Welcome to module_14 Homework’s documentation!

REST API main

async main.dbhealthchecker(db: AsyncSession = Depends(get_db))

Check if the database is configured correctly

Parameters:

db – AsyncSession: The database to check against

Returns:

A message indicating if the database is configured correctly

main.healthchecker()

Check if the application is running correctly

Returns:

A message indicating if the application is running correctly

REST API repository Contact

async repository.contact.create_contact(body: ContactCreateSchema, db: AsyncSession, user_id: int)

The create_contact function creates a new contact.

Parameters:

  • body – ContactCreateSchema: Validate the request body

  • db – AsyncSession: Access the database

  • user_id – int: Set the created_by field in the contact table

Returns:

A contact object, which is a mapped class

Doc-author:

Trelent

async repository.contact.delete_contact(contact_id: int, db: AsyncSession, user_id: int)
The delete_contact function deletes a contact from the database.
Args:

contact_id (int): The id of the contact to delete. db (AsyncSession): An async session object for interacting with the database. user_id (int): The id of the user who owns this contact.

Parameters:

  • contact_id – int: Specify which contact to delete

  • db – AsyncSession: Pass in the database session

  • user_id – int: Make sure that the user is only able to delete contacts they have created

Returns:

The contact object that was deleted

Doc-author:

Trelent

async repository.contact.edit_contact(contact_id: int, body: ContactEditSchema, db: AsyncSession, user_id: int)

The edit_contact function allows you to edit a contact.

Parameters:

  • contact_id – int: Identify the contact to edit

  • body – ContactEditSchema: Validate the request body

  • db – AsyncSession: Create a database session

  • user_id – int: Ensure that the user is only able to edit their own contacts

Returns:

The contact object that was modified

Doc-author:

Trelent

async repository.contact.get_birthday_list(db: AsyncSession, user_id: int)

The get_birthday_list function returns a list of contacts that have birthdays within the next 7 days.

Parameters:

  • db – AsyncSession: Create a database session

  • user_id – int: Specify the user_id of the person who created the contact

Returns:

A list of contacts who have birthdays in the next 7 days

Doc-author:

Trelent

async repository.contact.get_contact(contact_id: int, db: AsyncSession, user_id: int)

The get_contact function returns a contact object from the database.

Parameters:

  • contact_id – int: Filter the results by contact id

  • db – AsyncSession: Pass in the database session to use for querying

  • user_id – int: Ensure that the user is only able to access contacts they have created

Returns:

A contact object

Doc-author:

Trelent

async repository.contact.get_contact_by_mail(mail_query: str, db: AsyncSession, user_id: int)
The get_contact_by_mail function returns a list of contacts that match the mail_query string.
The function takes in three arguments:

  • mail_query: A string containing the email address to search for.

  • db: An async database session object from SQLAlchemy. This is used to execute queries against the database.

  • user_id: An integer representing which user created this contact.

Parameters:

  • mail_query – str: Query the database for a contact with an email that contains the mail_query string

  • db – AsyncSession: Pass the database session to the function

  • user_id – int: Check if the user is allowed to access the contact

Returns:

A list of contacts

Doc-author:

Trelent

async repository.contact.get_contact_by_name(name_query: str, db: AsyncSession, user_id: int)

The get_contact_by_name function returns a list of contacts that match the name query.

Parameters:

  • name_query – str: Pass in the name of the contact we want to search for

  • db – AsyncSession: Pass in the database connection to the function

  • user_id – int: Filter the query to only return contacts created by that user

Returns:

A list of dictionaries, where each dictionary is a row from the database

Doc-author:

Trelent

async repository.contact.get_contacts(limit: int, offset: int, db: AsyncSession, user_id: int)

The get_contacts function returns a list of contacts for the user.

Parameters:

  • limit – int: Limit the number of contacts returned

  • offset – int: Skip the first offset number of contacts

  • db – AsyncSession: Pass in the database session

  • user_id – int: Filter the contacts by user

Returns:

A list of scalars

Doc-author:

Trelent

REST API repository User

async repository.user.create_user(body: dict, db: AsyncSession)

The create_user function creates a new user in the database.

Parameters:

  • body – dict: Pass in the request body

  • db – AsyncSession: Pass the database session to the function

Returns:

A user object

Doc-author:

Trelent

async repository.user.delete_user(user_id: int, db: AsyncSession)
The delete_user function deletes a user from the database.
Args:

user_id (int): The id of the user to delete. db (AsyncSession): An async session object for interacting with the database.

Returns:

User: The deleted User object.

Parameters:

  • user_id – int: Specify the user to delete

  • db – AsyncSession: Pass the database connection to the function

Returns:

A user

Doc-author:

Trelent

async repository.user.edit_user(user_id: int, body: UserCreateSchema, db: AsyncSession)

The edit_user function allows you to edit a user.

Parameters:

  • user_id – int: Identify the user to be edited

  • body – UserCreateSchema: Validate the request body

  • db – AsyncSession: Pass in the database session

Returns:

The updated user object

Doc-author:

Trelent

async repository.user.find_user(email: str, db: AsyncSession) → User | None

The find_user function takes an email and a database session, and returns the user with that email if it exists. If no such user exists, None is returned.

Parameters:

  • email – str: Specify the type of the parameter

  • db – AsyncSession: Pass the database session to the function

Returns:

A user object or none

Doc-author:

Trelent

async repository.user.get_user(user_id: int, db: AsyncSession)

The get_user function returns a user object from the database.

Parameters:

  • user_id – int: Specify the user id of the user you want to get

  • db – AsyncSession: Pass the database connection to the function

Returns:

A user object

Doc-author:

Trelent

async repository.user.set_image(user: User, url: str, db: AsyncSession) → User

The set_image function sets the image attribute of a user.

Parameters:

  • user – User: Specify the user that will be updated

  • url – str: Set the image url of a user

  • db – AsyncSession: Pass the database session to the function

Returns:

A user object

Doc-author:

Trelent

async repository.user.update_otp(user: User, db: AsyncSession, otp: int | None = None)

The update_otp function updates the OTP for a user.

If no OTP is provided, it will generate one randomly.

Parameters:

  • user – User: Pass in the user object that is to be updated

  • db – AsyncSession: Pass the database connection to the function

  • otp – int | None: Set the otp to a specific value

Returns:

The otp

Doc-author:

Trelent

async repository.user.user_activation(user: User, db: AsyncSession, is_active: bool)

The user_activation function is used to activate or deactivate a user.

Parameters:

  • user – User: Pass the user object to the function

  • db – AsyncSession: Pass in the database session

  • is_active – bool: Set the user’s is_active attribute to true or false

Returns:

A boolean value

Doc-author:

Trelent

REST API routers Auth

async routers.auth.activate_user(user_mail: str = Query(PydanticUndefined), otp: int = Query(PydanticUndefined), db: AsyncSession = Depends(get_db), rl=Depends(limit_allowed))

The activate_user function is used to activate a user account. It takes in the email of the user and an OTP (one time password) that was sent to their email address. If both are correct, it will return a success message.

Parameters:

  • user_mail – str: Get the user’s email address from the request body

  • otp – int: Pass the otp to the activate_user function

  • db – AsyncSession: Pass the database session to the function

  • rl – Limit the number of requests per minute

Returns:

A user object

Doc-author:

Trelent

async routers.auth.login_for_access_token(form_data: OAuth2PasswordRequestForm, db: AsyncSession = Depends(get_db), rl=Depends(limit_allowed)) → Token

The login_for_access_token function is used to generate a new access token for the user. The function takes in an OAuth2PasswordRequestForm object, which contains the username and password of the user. It then uses these credentials to authenticate with our database, and if successful it will return a Token object containing the access_token that can be used by clients to make authenticated requests.

Parameters:

  • form_data – Annotated[OAuth2PasswordRequestForm: Get the data from the request body

  • Depends()] – Pass in the form_data from the request body

  • db – AsyncSession: Get the database connection

  • rl – Limit the number of requests per minute

Returns:

A token object, which is a pydantic model

Doc-author:

Trelent

async routers.auth.refresh_token(credentials: HTTPAuthorizationCredentials = Security(HTTPBearer), db: AsyncSession = Depends(get_db), rl=Depends(limit_allowed))

The refresh_token function is used to refresh the access token. The function will check if the user has a valid refresh token, and then return a new access token.

Parameters:

  • credentials – HTTPAuthorizationCredentials: Get the token from the request header

  • token (#) – str: Get the token from the request header

  • db – AsyncSession: Access the database

  • rl – Limit the number of requests that can be made to this endpoint

Returns:

A new access token, which is then used to authenticate the user and provide access to protected resources

Doc-author:

Trelent

async routers.auth.send_mail(token: str = Depends(OAuth2PasswordBearer), db: AsyncSession = Depends(get_db), rl=Depends(limit_allowed))

The send_mail function sends an email to the user’s email address. The function takes in a token, which is used to authenticate the user. It also takes in a database session and rate limit object as dependencies.

Parameters:

  • token – str: Get the token from the header

  • current_user (#) – auth.DefaultUser: Get the current user

  • db – AsyncSession: Get the database session

  • rl – Limit the number of requests a user can make

Returns:

A status message

Doc-author:

Trelent

async routers.auth.signup(body: UserCreateSchema, db: AsyncSession = Depends(get_db), rl=Depends(limit_allowed))

The signup function creates a new user in the database. It takes a UserCreateSchema as input, which is validated by FastAPI. The function then uses the auth module to create the user and return it.

Parameters:

  • body – UserCreateSchema: Validate the request body

  • db – AsyncSession: Pass the database session to the function

  • rl – Limit the number of requests that can be made to this endpoint

Returns:

A user object

Doc-author:

Trelent

async routers.auth.user_path(token: str = Depends(OAuth2PasswordBearer), db: AsyncSession = Depends(get_db), rl=Depends(limit_allowed))

The secret function is a protected endpoint that returns the current user’s username.

Parameters:

  • token – str: Get the token from the request

  • current_user (#) – auth.DefaultUser: Get the current user

  • db – AsyncSession: Get the database session

  • rl – Limit the number of requests per user

Returns:

The user object

Doc-author:

Trelent

REST API routers Contact

async routers.contact.add_contact(body: ContactCreateSchema, db: AsyncSession = Depends(get_db), token: str = Depends(OAuth2PasswordBearer))
The add_contact function creates a new contact for the user.

The function takes in a ContactCreateSchema object, which is validated by pydantic. It also takes in an optional database session and token string, which are used to authenticate the user.

Parameters:

  • body – ContactCreateSchema: Validate the request body

  • db – AsyncSession: Pass the database session to the function

  • token – str: Get the user’s token from the request header

Returns:

A contact object

Doc-author:

Trelent

async routers.contact.birthday_list(db: AsyncSession = Depends(get_db), token: str = Depends(OAuth2PasswordBearer))

The birthday_list function returns a list of contacts with birthdays in the current month.

Parameters:

  • db – AsyncSession: Get the database connection

  • token – str: Get the user id from the token

Returns:

A list of contacts, but the schema for a contact is not defined

Doc-author:

Trelent

async routers.contact.delete_contact(id_: int, db: AsyncSession = Depends(get_db), token: str = Depends(OAuth2PasswordBearer))

The delete_contact function deletes a contact from the database.

Parameters:

  • id – int: Get the id of the contact that is going to be deleted

  • db – AsyncSession: Pass the database session to the function

  • token – str: Get the token from the request header

Returns:

The deleted contact

Doc-author:

Trelent

async routers.contact.edit_contact(id_: int, body: ContactEditSchema, db: AsyncSession = Depends(get_db), token: str = Depends(OAuth2PasswordBearer))

The edit_contact function allows a user to edit an existing contact.

Parameters:

  • id – int: Identify the contact to be edited

  • body – ContactEditSchema: Validate the request body

  • db – AsyncSession: Pass the database connection to the function

  • token – str: Get the current user from the database

Returns:

A contact object

Doc-author:

Trelent

async routers.contact.get_contact(id_: int, db: AsyncSession = Depends(get_db), token: str = Depends(OAuth2PasswordBearer))
The get_contact function returns a contact by id.

If the contact does not exist, it will return an HTTP 303 See Other status code.

Parameters:

  • id – int: Get the contact id

  • db – AsyncSession: Pass the database connection to the function

  • token – str: Validate the token

Returns:

A contact object, or status code if no contact is found

Doc-author:

Trelent

async routers.contact.list_contacts(limit: int = Query(10), offset: int = Query(0), db: AsyncSession = Depends(get_db), token: str = Depends(OAuth2PasswordBearer))
The list_contacts function returns a list of contacts for the user.

The limit and offset parameters are used to paginate the results.

Parameters:

  • request (#) – Request: Get the request object

  • limit – int: Limit the number of contacts returned

  • ge – Specify the minimum value of a parameter

  • le – Limit the number of contacts returned to 500

  • offset – int: Get the offset of the contacts to be returned

  • ge – Set a minimum value for the limit parameter

  • db – AsyncSession: Pass the database session to the function

  • token – str: Get the user id from the token

Returns:

A list of contacts

Doc-author:

Trelent

async routers.contact.search_mail(query: str | None = None, db: AsyncSession = Depends(get_db), token: str = Depends(OAuth2PasswordBearer))

The search_mail function searches for contacts by email.

Parameters:

  • query – str | None: Get the query string from the url

  • db – AsyncSession: Get the database session

  • token – str: Get the user id from the token

Returns:

An array of contacts

Doc-author:

Trelent

async routers.contact.search_name(query: str | None = None, db: AsyncSession = Depends(get_db), token: str = Depends(OAuth2PasswordBearer))

The search_name function searches for contacts by name.

Parameters:

  • query – str | None: Get the query parameter from the url

  • db – AsyncSession: Get the database connection

  • token – str: Get the current user from the database

Returns:

A list of contacts

Doc-author:

Trelent

REST API schemas Contact

class schemas.contact.ContactCreateSchema(*, first_name: str, last_name: str, email: EmailStr, birth_date: date)

Bases: BaseModel

birth_date: date
email: EmailStr
first_name: str
last_name: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'birth_date': FieldInfo(annotation=date, required=True), 'email': FieldInfo(annotation=EmailStr, required=True), 'first_name': FieldInfo(annotation=str, required=True), 'last_name': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class schemas.contact.ContactEditSchema(*, first_name: str | None = None, last_name: str | None = None, email: EmailStr | None = None, birth_date: date | None = None)

Bases: ContactCreateSchema

birth_date: date | None
email: EmailStr | None
first_name: str | None
last_name: str | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'birth_date': FieldInfo(annotation=Union[date, NoneType], required=False), 'email': FieldInfo(annotation=Union[EmailStr, NoneType], required=False), 'first_name': FieldInfo(annotation=Union[str, NoneType], required=False), 'last_name': FieldInfo(annotation=Union[str, NoneType], required=False)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class schemas.contact.ContactViewSchema(*, first_name: str, last_name: str, email: EmailStr, birth_date: date, id: int, created_by: int, created_at: datetime, modified_at: datetime)

Bases: ContactCreateSchema

class Config

Bases: object

from_attributes = True
created_at: datetime
created_by: int
id: int
model_config: ClassVar[ConfigDict] = {'from_attributes': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'birth_date': FieldInfo(annotation=date, required=True), 'created_at': FieldInfo(annotation=datetime, required=True), 'created_by': FieldInfo(annotation=int, required=True), 'email': FieldInfo(annotation=EmailStr, required=True), 'first_name': FieldInfo(annotation=str, required=True), 'id': FieldInfo(annotation=int, required=True), 'last_name': FieldInfo(annotation=str, required=True), 'modified_at': FieldInfo(annotation=datetime, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

modified_at: datetime

REST API schemas Auth

class schemas.auth.Token(*, access_token: str, refresh_token: str, token_type: str)

Bases: BaseModel

access_token: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'access_token': FieldInfo(annotation=str, required=True), 'refresh_token': FieldInfo(annotation=str, required=True), 'token_type': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

refresh_token: str
token_type: str
class schemas.auth.TokenData(*, email: EmailStr)

Bases: BaseModel

email: EmailStr
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'email': FieldInfo(annotation=EmailStr, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class schemas.auth.User(*, email: EmailStr)

Bases: BaseModel

email: EmailStr
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'email': FieldInfo(annotation=EmailStr, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class schemas.auth.UserInDB(*, email: EmailStr, hashed_password: str)

Bases: User

hashed_password: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'email': FieldInfo(annotation=EmailStr, required=True), 'hashed_password': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

REST API schemas User

class schemas.user.Token(*, access_token: str, token_type: str = 'bearer')

Bases: BaseModel

access_token: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'access_token': FieldInfo(annotation=str, required=True), 'token_type': FieldInfo(annotation=str, required=False, default='bearer')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

token_type: str
class schemas.user.UserCreateSchema(*, email: EmailStr, passwd: str)

Bases: BaseModel

email: EmailStr
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'email': FieldInfo(annotation=EmailStr, required=True), 'passwd': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

passwd: str
class schemas.user.UserViewSchema(*, email: EmailStr, passwd: str, id: int, created_at: datetime, modified_at: datetime)

Bases: UserCreateSchema

created_at: datetime
id: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'created_at': FieldInfo(annotation=datetime, required=True), 'email': FieldInfo(annotation=EmailStr, required=True), 'id': FieldInfo(annotation=int, required=True), 'modified_at': FieldInfo(annotation=datetime, required=True), 'passwd': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

modified_at: datetime

REST API service Auth

async service.auth.activate_user(user_mail: str, otp: int, db: AsyncSession)
The activate_user function is used to activate a user account.
It takes the following parameters:

  • user_mail (str): The email of the user to be activated.

  • otp (int): The one time password sent to the users email address for activation purposes.

Parameters:

  • user_mail – str: Find the user in the database

  • otp – int: Check if the otp sent by the user is equal to the one in our database

  • db – AsyncSession: Pass the database session to the function

Returns:

The user object

Doc-author:

Trelent

async service.auth.authenticate_user(db, email: str, password: str) → User | None

The authenticate_user function takes a database connection and an email address and password. It returns the user object if the credentials are valid, otherwise it returns None.

Parameters:

  • db – Access the database

  • email – str: Get the email of the user

  • password – str: Get the password from the user

Returns:

A dbuser object if the user is authenticated, or none otherwise

Doc-author:

Trelent

service.auth.create_access_token(data: dict, expires_delta: timedelta | None = None)
The create_access_token function creates a JWT token with the following claims:

  • exp: expiration time of the token. The default is 15 minutes after creation.

  • iat: issued at time, in UTC format.

  • scope: access_token (this is not used by our application but can be useful for other applications)

Parameters:

  • data – dict: Store the data that we want to encode in our jwt

  • expires_delta – timedelta | None: Set the expiration time of the token

Returns:

A string, which is the encoded jwt

Doc-author:

Trelent

service.auth.create_refresh_token(data: dict, expires_delta: timedelta | None = None)
The create_refresh_token function creates a refresh token for the user.
Args:

data (dict): The data to be encoded in the JWT. This should include at least a username and an email address, but can also include other information such as roles or permissions. expires_delta (timedelta | None): The amount of time until this token expires, defaults to 1000 minutes if not specified.

Parameters:

  • data – dict: Pass the user’s id to the function

  • expires_delta – timedelta | None: Set the expiration time of the token

Returns:

A string

Doc-author:

Trelent

async service.auth.create_user(body: UserCreateSchema, db: AsyncSession)
The create_user function creates a new user in the database.

It takes in a UserCreateSchema object, which is validated by pydantic and then converted into a dictionary. The function then checks if the email already exists in the database, and if it does not exist, hashes the password using Argon2 hashing algorithm with salt value of 16 bytes. Then it creates an OTP for that user and sends an email to that user’s registered email address.

Parameters:

  • body – UserCreateSchema: Validate the data that is being passed in

  • db – AsyncSession: Pass the database session to the function

Returns:

A user object

Doc-author:

Trelent

async service.auth.decode_refresh_token(refresh_token: str)

The decode_refresh_token function decodes the refresh token and returns the email address of the user. If it fails to decode, it raises an HTTPException with a 401 status code.

Parameters:

refresh_token – str: Pass the refresh token to the function

Returns:

The email of the user

Doc-author:

Trelent

async service.auth.generate_token(db: AsyncSession, form_data: OAuth2PasswordRequestForm) → Token

The generate_token function is used to generate a new access token and refresh token.

Parameters:

  • db – AsyncSession: Pass in the database session

  • form_data – Annotated[OAuth2PasswordRequestForm: Validate the data sent in the request body

  • Depends()] – Inject dependencies into the function

Returns:

A token object

Doc-author:

Trelent

async service.auth.get_current_active_user(current_user: User)

The get_current_active_user function returns the current active user.

Parameters:

  • current_user – Annotated[User: Get the current user from the database

  • Depends(get_current_user)] – Get the current user

Returns:

The current user, which is the user that has been authenticated

Doc-author:

Trelent

async service.auth.get_current_user(token: str, db: AsyncSession) → User

The get_current_user function is a dependency that will be called by the FastAPI framework to retrieve the current user.

Parameters:

  • token – Annotated[str: Get the token from the request header

  • Depends(oauth2_scheme)] – Check if the user is logged in

  • db – AsyncSession: Connect to the database

Returns:

The user object that is associated with the token

Doc-author:

Trelent

service.auth.get_password_hash(password)

The get_password_hash function takes a password as input and returns the hash of that password.

Parameters:

password – Get the password from the user

Returns:

A hash of the password

Doc-author:

Trelent

async service.auth.get_user(db: AsyncSession, email: str) → User | None

The get_user function is used to retrieve a user from the database.

Parameters:

  • db – AsyncSession: Pass the database session to the function

  • email – str: Find the user in the database

Returns:

A dbuser object or none

Doc-author:

Trelent

async service.auth.hash_pwd(pwd: str, salt=None) → tuple[str, str]

The hash_pwd function takes a password and an optional salt. If no salt is provided, it will generate one using bcrypt’s gensalt function. It then hashes the password with the given or generated salt using bcrypt’s hashpw function. The hashed_pwd and the used salt are returned as a tuple of strings.

Parameters:

  • pwd – str: Pass in the password that is to be hashed

  • salt – Generate a unique hash for each password

Returns:

A tuple of strings

Doc-author:

Trelent

async service.auth.refresh_user_token(token: str, db: AsyncSession) → Token

The refresh_user_token function is used to refresh the access token. It takes a valid refresh token and returns a new access token.

Parameters:

  • token – Annotated[str: Get the token from the request header

  • Depends(oauth2_scheme)] – Validate the token

  • db – AsyncSession: Pass the database session to the function

Returns:

A token object

Doc-author:

Trelent

service.auth.verify_password(plain_password, hashed_password)

The verify_password function takes a plain-text password and the hashed version of that password, and returns True if they match, False otherwise. This is used to verify that the user’s login attempt is valid.

Parameters:

  • plain_password – Pass in the password that was entered by the user

  • hashed_password – Compare the hashed password in the database to a plain text password

Returns:

A boolean value, true or false

Doc-author:

Trelent

REST API service Avatar

REST API service Contact

REST API service Emails

service.emails.send_email(subject, message, to_email)

The send_email function sends an email to the user with a link to reset their password.

Parameters:

  • subject – Set the subject of the email

  • message – Pass the message that will be sent to the user

  • to_email – Specify the email address of the recipient

Returns:

None

Doc-author:

Trelent

REST API service Rate Limiter

class service.rate_limiter.RateLimiter(*args, **kwargs)

Bases: object

is_allowed(client_id)

The is_allowed function takes in a client_id and returns True if the client is allowed to make another request, and False otherwise. The function keeps track of how many requests each client has made within a window of time. If the number of requests exceeds max_requests within that window, then the function will return False for that client until enough time has passed.

Parameters:

  • self – Allow an instance of the class to access its own attributes and methods

  • client_id – Identify the client making the request

Returns:

True or false

Doc-author:

Trelent

class service.rate_limiter.RateLimiterMeta

Bases: type

The Singleton class can be implemented in different ways in Python. Some possible methods include: base class, decorator, metaclass. We will use the metaclass because it is best suited for this purpose.

service.rate_limiter.limit_allowed(request: Request) → bool
The limit_allowed function is a rate limiter that limits the number of requests per client.

It uses the RateLimiter class to accomplish this task. The function takes in a request object, and returns True if the client has not exceeded their limit, or raises an HTTPException with status code 429 (Too Many Requests) if they have.

Parameters:

request – Request: Get the client id from the request

Returns:

True or raises an httpexception

Doc-author:

Trelent

Indices and tables