Welcome to Contacts Rest API’s documentation!

REST API main

async main.add_process_time_header(request: Request, call_next)

This middleware adds a header to the response with the time taken to process the request.

Parameters: request (Request): The incoming HTTP request. call_next (Callable): A callback function that represents the next middleware or route handler in the application.

Returns: Response: The response object with an additional header “X-Process-Time” containing the time taken to process the request.

Raises: None

async main.ban_ips(request: Request, call_next: Callable)

This middleware checks if the client’s IP address is in the list of banned IPs. If it is, it returns a 403 Forbidden response with a message indicating that the client is banned. Otherwise, it calls the next middleware or route handler in the application.

Parameters: request (Request): The incoming HTTP request. call_next (Callable): A callback function that represents the next middleware or route handler in the application.

Returns: Response: If the client’s IP address is not banned, the response returned by the next middleware or route handler is returned. If the client’s IP address is banned, a JSONResponse with a 403 Forbidden status code and a message indicating that the client is banned is returned.

Raises: None

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

This function is a route handler for the “/api/healthchecker” endpoint. It checks the connection to the database and returns a success message if the connection is established.

Parameters: - db (AsyncSession): A dependency representing the database session. It is obtained using the get_db function.

Returns: - dict: A dictionary containing a message indicating that the connection to the database is established.

Raises: - HTTPException: If there is an error connecting to the database, an HTTPException with a status code of 500 and a message indicating the error is raised.

async main.lifespan(app: FastAPI)

Initializes the FastAPILimiter middleware with the provided Redis client. This middleware is used to limit the number of requests a client can make within a certain time period.

Parameters: app (FastAPI): The FastAPI application instance.

Returns: None

Yields: None

Raises: None

This function is called during the lifespan of the FastAPI application. It initializes the FastAPILimiter middleware with the provided Redis client, which is used to limit the number of requests a client can make within a certain time period. After the initialization, it yields control back to the application.

main.read_root(request: Request)

This function is a route handler for the “/” endpoint. It renders the home page of the application using the Jinja2 templating engine.

Parameters: - request (Request): The incoming HTTP request object.

Returns: - Response: A response object containing the rendered HTML content of the “index.html” template.

async main.redis_healthchecker(key='test', value='test', redis_client: Redis = Depends(get_redis_client))

This function is a route handler for the “/api/redis_healthchecker” endpoint. It checks the connection to the Redis database and returns a success message if the connection is established.

Parameters: - key (str): A string representing the key to be set in the Redis database. Default is “test”. - value (str): A string representing the value to be set for the given key in the Redis database. Default is “test”. - redis_client (Redis): A dependency representing the Redis client. It is obtained using the get_redis_client function.

Returns: - dict: A dictionary containing a message indicating that the connection to the Redis database is established.

Raises: - HTTPException: If there is an error connecting to the Redis database, an HTTPException with a status code of 500 and a message indicating the error is raised.

REST API Database

async src.database.db.get_db()

Asynchronously creates a database session using SQLAlchemy’s async_sessionmaker.

Parameters: None

Returns: async_generator: An asynchronous generator yielding a SQLAlchemy session.

Raises: HTTPException: If a SQLAlchemyError occurs during session creation, a 500 error is raised.

async src.database.db.get_redis_client()

Asynchronously creates a Redis client using aioredis.

Parameters: None

Returns: aioredis.Redis: An asynchronous Redis client.

Raises: HTTPException: If a RedisError occurs during client creation, a 500 error is raised.

REST API repository Users

async src.repository.users.confirmed_email(email: str, db: AsyncSession)

Confirms the email for the given user in the database.

Args: - email (str): The email address of the user to confirm. - db (AsyncSession): The database session to use for the update.

Returns: - None: This function does not return a value.

Raises: - None: This function does not raise any exceptions.

This function confirms the email for the given user in the database. It retrieves the user object from the database using the provided email address, sets the confirmed attribute of the user to True, and then commits the changes to the database. Finally, it refreshes the user object to ensure that the changes are reflected in the returned object.

async src.repository.users.create_user(body: UserSchema, db: AsyncSession = Depends(get_db))

Creates a new user in the database.

Args: - body (UserSchema): The user schema containing the user’s data. - db (AsyncSession, optional): The database session to use for the query. Defaults to Depends(get_db).

Returns: - User: The newly created user object.

Raises: - HTTPException: If an error occurs while creating the user.

This function first generates an avatar for the user using the Gravatar library. If an error occurs during this process, it raises an HTTPException with a status code of 500 (Internal Server Error). It then creates a new User object using the data from the provided UserSchema, sets the avatar, and adds it to the database session. Finally, it commits the changes to the database and refreshes the user object before returning it.

async src.repository.users.get_user_by_email(email: str, db: AsyncSession = Depends(get_db))

Retrieves a user from the database by their email address.

Args: - email (str): The email address of the user to retrieve. - db (AsyncSession, optional): The database session to use for the query. Defaults to Depends(get_db).

Returns: - User | None: The user object if found, otherwise None.

Raises: - HTTPException: If an error occurs while retrieving the user.

async src.repository.users.update_avatar(email, url: str, db: AsyncSession) → User

Updates the avatar URL for the given user in the database.

Args: - email (str): The email address of the user to update the avatar for. - url (str): The new avatar URL to set for the user. - db (AsyncSession): The database session to use for the update.

Returns: - User: The updated user object with the new avatar URL.

Raises: - None: This function does not raise any exceptions.

This function updates the avatar URL for the given user in the database. It retrieves the user object from the database using the provided email address, sets the avatar attribute of the user to the new URL, and then commits the changes to the database. Finally, it refreshes the user object to ensure that the changes are reflected in the returned object.

async src.repository.users.update_password(user: User, new_password: str, db: AsyncSession)

Updates the password for the given user in the database.

Args: - user (User): The user object to update the password for. - new_password (str): The new password to set for the user. - db (AsyncSession): The database session to use for the update.

Returns: - User: The updated user object with the new password.

Raises: - None: This function does not raise any exceptions.

This function updates the password for the given user in the database. It retrieves the user object from the database using the provided user object, sets the password attribute of the user to the new password, and then commits the changes to the database. Finally, it refreshes the user object to ensure that the changes are reflected in the returned object.

async src.repository.users.update_token(user: User, refresh_token: str | None, db: AsyncSession)

Updates the refresh token for the given user in the database.

Args: - user (User): The user object to update the refresh token for. - refresh_token (str | None): The new refresh token to set for the user, or None to remove the refresh token. - db (AsyncSession): The database session to use for the update.

Returns: - None: This function does not return a value.

Raises: - None: This function does not raise any exceptions.

This function updates the refresh token for the given user in the database. If refresh_token is None, the refresh token for the user is removed. The changes are then committed to the database.

REST API repository Contacts

async src.repository.contacts.create_contact(body: ContactSchema, db: AsyncSession, user: User)

Creates a new contact for the specified user.

Args: - body (ContactSchema): The schema containing the details of the new contact. - db (AsyncSession): The database session. - user (User): The user for whom the contact is being created.

Raises: - HTTPException: If a contact with the same first name, last name, and email already exists.

Returns: - Contact: The newly created contact object.

async src.repository.contacts.delete_contact(contact_id: int, db: AsyncSession, user: User)

Deletes a specific contact by its id for the specified user.

Args: - contact_id (int): The id of the contact to delete. - db (AsyncSession): The database session. - user (User): The user for whom the contact is being deleted.

Returns: - Contact or None: The specific contact object for the specified user and contact_id, or None if not found.

async src.repository.contacts.get_birthdays(db: AsyncSession, user: User)

Retrieves a list of contacts for the specified user with birthdays within the next 7 days.

Args: - db (AsyncSession): The database session. - user (User): The user for whom the contacts are being retrieved.

Returns: - list: A list of Contact objects for the specified user with birthdays within the next 7 days.

Raises: - HTTPException: If no contacts are found with the given birthday and user.

async src.repository.contacts.get_contact(contact_id: int, db: AsyncSession, user: User)

Retrieves a specific contact by its id for the specified user.

Args: - contact_id (int): The id of the contact to retrieve. - db (AsyncSession): The database session. - user (User): The user for whom the contact is being retrieved.

Returns: - Contact or None: The specific contact object for the specified user and contact_id, or None if not found.

async src.repository.contacts.get_contact_by_birthday(contact_birthday: Date, db: AsyncSession, user: User)

Retrieves a list of contacts for the specified user with the given birthday.

Args: - contact_birthday (Date): The birthday of the contact to retrieve. - db (AsyncSession): The database session. - user (User): The user for whom the contact is being retrieved.

Returns: - list: A list of Contact objects for the specified user and birthday.

Raises: - HTTPException: If no contacts are found with the given birthday and user.

async src.repository.contacts.get_contact_by_email(contact_email: str, db: AsyncSession, user: User)

Retrieves a specific contact by its email for the specified user.

Args: - contact_email (str): The email of the contact to retrieve. - db (AsyncSession): The database session. - user (User): The user for whom the contact is being retrieved.

Returns: - Contact or None: The specific contact object for the specified user and contact_email, or None if not found.

async src.repository.contacts.get_contact_by_last_name(contact_last_name: str, db: AsyncSession, user: User) → list

Retrieves a list of contacts for the specified user with the given last name.

Args: - contact_last_name (str): The last name of the contact to retrieve. - db (AsyncSession): The database session. - user (User): The user for whom the contact is being retrieved.

Returns: - list: A list of Contact objects for the specified user and last name.

async src.repository.contacts.get_contacts(limit: int, offset: int, db: AsyncSession, user: User)

Retrieves a list of contacts for the specified user within the specified limit and offset.

Args: - limit (int): The maximum number of contacts to retrieve. - offset (int): The index from which to start retrieving contacts. - db (AsyncSession): The database session. - user (User): The user for whom the contacts are being retrieved.

Returns: - list: A list of Contact objects for the specified user within the specified limit and offset.

async src.repository.contacts.update_contact(contact_id: int, body: ContactUpdateSchema, db: AsyncSession, user: User) → Contact

Updates a specific contact by its id for the specified user.

Args: - contact_id (int): The id of the contact to update. - body (ContactUpdateSchema): The schema containing the details of the updated contact. - db (AsyncSession): The database session. - user (User): The user for whom the contact is being updated.

Returns: - Contact: The updated contact object.

REST API routes Users

async src.routes.users.get_current_user(user: User = Depends(get_current_user))

Retrieves the current authenticated user’s information.

Parameters: - user (User): The authenticated user object obtained from the auth_service.

Returns: - UserResponseSchema: The current authenticated user’s information in the specified response model format.

This endpoint is protected by a rate limiter that allows no more than 2 requests per 10 seconds.

async src.routes.users.update_avatar_user(file: UploadFile = File(PydanticUndefined), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db))

Updates the avatar of the current authenticated user.

Parameters: - file (UploadFile): The file to be uploaded as the new avatar. - current_user (User): The authenticated user object obtained from the auth_service. - db (AsyncSession): The database session to be used for the operation.

Returns: - UserDb: The updated user object with the new avatar URL.

This method uses Cloudinary to upload the new avatar file and update the user’s avatar URL in the database.

REST API routes Contacts

async src.routes.contacts.create_contact(body: ContactSchema, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user))

Create a new contact.

Parameters: - body (ContactSchema): The data of the new contact to be created. - db (AsyncSession): The database session to use for the operation. - user (User): The authenticated user making the request.

Returns: - ContactResponseSchema: The newly created contact data.

async src.routes.contacts.delete_contact(contact_id: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user))

Delete a contact by its ID.

Parameters: - contact_id (int): The ID of the contact to delete. - db (AsyncSession): The database session to use for the operation. - user (User): The authenticated user making the request.

Returns: - ContactResponseSchema: The deleted contact data. - HTTPException: If the contact with the specified ID is not found.

async src.routes.contacts.get_birthdays(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user))

Get all contacts with birthdays within 7 days.

Parameters: - db (AsyncSession): The database session to use for the operation. - user (User): The authenticated user making the request.

Returns: - list[ContactResponseSchema]: A list of ContactResponseSchema objects representing the contacts with birthdays within 7 days.

async src.routes.contacts.get_contact(contact_id: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user))

Get a specific contact by its ID.

Parameters: - contact_id (int): The ID of the contact to retrieve. - db (AsyncSession): The database session to use for the operation. - user (User): The authenticated user making the request.

Returns: - ContactResponseSchema: The data of the specific contact. - HTTPException: If the contact with the specified ID is not found.

async src.routes.contacts.get_contact_by_birthday(contact_birthday: str, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user))

Get all contacts with birthdays on the specified date.

Parameters: - contact_birthday (str): The date in the format ‘YYYY-MM-DD’ to search for contacts with birthdays. - db (AsyncSession): The database session to use for the operation. - user (User): The authenticated user making the request.

Returns: - list[ContactResponseSchema]: A list of ContactResponseSchema objects representing the contacts with birthdays on the specified date.

async src.routes.contacts.get_contact_by_email(contact_email: str, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user))

Get a contact by their email address.

Parameters: - contact_email (str): The email address of the contact to search for. - db (AsyncSession): The database session to use for the operation. - user (User): The authenticated user making the request.

Returns: - ContactResponseSchema: A ContactResponseSchema object representing the contact with the specified email address. - HTTPException: If the contact with the specified email address is not found.

async src.routes.contacts.get_contact_by_last_name(contact_last_name: str, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user))

Get all contacts with the specified last name.

Parameters: - contact_last_name (str): The last name to search for in the contacts. - db (AsyncSession): The database session to use for the operation. - user (User): The authenticated user making the request.

Returns: - list[ContactResponseSchema]: A list of ContactResponseSchema objects representing the contacts with the specified last name.

async src.routes.contacts.get_contacts(limit: int = Query(10), offset: int = Query(0), db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user))

Get a list of contacts.

Parameters: - limit (int): The maximum number of contacts to retrieve. Default is 10, minimum is 1, and maximum is 500. - offset (int): The index of the first contact to retrieve. Default is 0. - db (AsyncSession): The database session to use for the operation. - user (User): The authenticated user making the request.

Returns: - list[ContactResponseSchema]: A list of ContactResponseSchema objects representing the contacts.

async src.routes.contacts.update_contact(body: ContactUpdateSchema, contact_id: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user))

Update an existing contact.

Parameters: - body (ContactUpdateSchema): The updated contact data. - contact_id (int): The ID of the contact to update. - db (AsyncSession): The database session to use for the operation. - user (User): The authenticated user making the request.

Returns: - ContactResponseSchema: The updated contact data. - HTTPException: If the contact with the specified ID is not found.

REST API routes Auth

async src.routes.auth.confirmed_email(token: str, db: AsyncSession = Depends(get_db))

Confirm the user’s email address.

Parameters: - token (str): The token sent to the user’s email address for confirmation. - db (AsyncSession): The database session.

Returns: - dict: A dictionary containing a message indicating whether the email address has been confirmed.

Raises: - HTTPException: If the token is invalid or expired.

async src.routes.auth.confirmed_reset_password_email(token: str, request: Request, db: AsyncSession = Depends(get_db))

Retrieves the reset password form for the user with the given token.

Parameters: - token (str): The token sent to the user’s email address for resetting the password. - request (Request): The HTTP request object. - db (AsyncSession): The database session.

Returns: - TemplateResponse: A Flask template response containing the reset password form.

Raises: - HTTPException: If the token is invalid or expired.

async src.routes.auth.get_cash(key: str, redis_client: Redis = Depends(get_redis_client), user: User = Depends(get_current_user))

Retrieves the value associated with the given key from the Redis cache.

Parameters: - key (str): The key to be retrieved from the cache. - redis_client (Redis): The Redis client used to interact with the cache. - user (User): The authenticated user making the request.

Returns: - dict: A dictionary containing the key and its associated value in the cache.

Raises: - HTTPException: If the user is not authenticated.

async src.routes.auth.login(body: OAuth2PasswordRequestForm = Depends(NoneType), db: AsyncSession = Depends(get_db))

Logs in a user and returns a JWT token.

Parameters: - body (OAuth2PasswordRequestForm): The user credentials. - db (AsyncSession): The database session.

Returns: - TokenSchema: A dictionary containing the access token, refresh token, and token type.

Raises: - HTTPException: If the user does not exist, if the user is not confirmed, or if the password is incorrect.

async src.routes.auth.refresh_token(credentials: HTTPAuthorizationCredentials = Security(HTTPBearer), db: AsyncSession = Depends(get_db))

Refresh the user’s access token.

Parameters: - credentials (HTTPAuthorizationCredentials): The HTTP Authorization header containing the refresh token. - db (AsyncSession): The database session.

Returns: - TokenSchema: A dictionary containing the new access token, refresh token, and token type.

Raises: - HTTPException: If the refresh token is invalid or expired.

async src.routes.auth.request_email(body: RequestEmail, background_tasks: BackgroundTasks, request: Request, db: AsyncSession = Depends(get_db))

Request a confirmation email for the user’s email address.

Parameters: - body (RequestEmail): A dictionary containing the user’s email address. - background_tasks (BackgroundTasks): A task manager for asynchronous tasks. - request (Request): The HTTP request object. - db (AsyncSession): The database session.

Returns: - dict: A dictionary containing a message indicating whether the email address has been confirmed.

Raises: - HTTPException: If the user already has a confirmed email address.

async src.routes.auth.request_password_reset(body: RequestEmail, background_tasks: BackgroundTasks, request: Request, db: AsyncSession = Depends(get_db))

Request a password reset email for the user’s email address.

Parameters: - body (RequestEmail): A dictionary containing the user’s email address. - background_tasks (BackgroundTasks): A task manager for asynchronous tasks. - request (Request): The HTTP request object. - db (AsyncSession): The database session.

Returns: - dict: A dictionary containing a message indicating whether the email address has been confirmed.

Raises: - HTTPException: If the user already has a confirmed email address.

async src.routes.auth.reset_password(token: str = Form(PydanticUndefined), new_password: str = Form(PydanticUndefined), db: AsyncSession = Depends(get_db))

Reset the user’s password using a token sent to their email address.

Parameters: - token (str): The token sent to the user’s email address for resetting the password. - new_password (str): The new password to be set for the user. - db (AsyncSession): The database session.

Returns: - JSONResponse: A JSON response containing a message indicating the success of the password reset.

Raises: - HTTPException: If the token is invalid or expired, or if the user is not found.

async src.routes.auth.set_cash(key: str, value: str, redis_client: Redis = Depends(get_redis_client), user: User = Depends(get_current_user))

Sets a key-value pair in the Redis cache.

Parameters: - key (str): The key to be set in the cache. - value (str): The value to be associated with the key in the cache. - redis_client (Redis): The Redis client used to interact with the cache. - user (User): The authenticated user making the request.

Returns: - None: This method does not return a value.

Raises: - HTTPException: If the user is not authenticated.

async src.routes.auth.signup(body: UserSchema, bt: BackgroundTasks, request: Request, db: AsyncSession = Depends(get_db))

Sign up a new user.

Parameters: - body (UserSchema): The user data to be created. - bt (BackgroundTasks): A task manager for asynchronous tasks. - request (Request): The HTTP request object. - db (AsyncSession): The database session.

Returns: - UserResponseSchema: The newly created user object.

Raises: - HTTPException: If the user already exists.

REST API service Auth

class src.services.auth.Auth

Bases: object

async create_access_token(data: dict, expires_delta: float | None = None)

Generates an access token for the user.

Args:

data (dict): A dictionary containing user-specific data. expires_delta (Optional[float], optional): The time in seconds that the token should be valid for. Defaults to 15 minutes.

Returns:

str: The generated access token.

Raises:

Exception: If the token generation process fails.

Note:

The token is generated using the JWT library with the provided SECRET_KEY and ALGORITHM.

create_email_token(data: dict)

Generates an email token for the user.

Args:

data (dict): A dictionary containing user-specific data.

Returns:

str: The generated email token.

Note:

The token is generated using the JWT library with the provided SECRET_KEY and ALGORITHM. The token has a validity of 1 day.

async create_refresh_token(data: dict, expires_delta: float | None = None)

Generates a refresh token for the user.

Args:

data (dict): A dictionary containing user-specific data. expires_delta (Optional[float], optional): The time in seconds that the token should be valid for. Defaults to 7 days.

Returns:

str: The generated refresh token.

Raises:

Exception: If the token generation process fails.

Note:

The token is generated using the JWT library with the provided SECRET_KEY and ALGORITHM.

async decode_refresh_token(refresh_token: str)

Decodes a refresh token to retrieve the user’s email.

Args:

refresh_token (str): The refresh token to be decoded.

Returns:

str: The user’s email if the token is valid and has a valid scope.

Raises:

HTTPException: If the token cannot be decoded or has an invalid scope.

Note:

The token is decoded using the JWT library with the provided SECRET_KEY and ALGORITHM.

async get_current_user(token: str = Depends(OAuth2PasswordBearer), db: AsyncSession = Depends(get_db), redis_client: Redis = Depends(get_redis_client))

Retrieves the current user from the token provided.

Args:

token (str): The token to be decoded and validated. db (AsyncSession, optional): The database session to be used for retrieving user data. Defaults to Depends(get_db). redis_client (Redis, optional): The Redis client to be used for caching user data. Defaults to Depends(get_redis_client).

Returns:

User: The current user object if the token is valid and has a valid scope.

Raises:

HTTPException: If the token cannot be decoded or has an invalid scope.

Note:

The token is decoded using the JWT library with the provided SECRET_KEY and ALGORITHM. If the user data is found in Redis, it is returned directly. Otherwise, the user data is retrieved from the database and cached in Redis before being returned.

async get_email_from_token(token: str)

Decodes a token to retrieve the user’s email.

Args:

token (str): The token to be decoded.

Returns:

str: The user’s email if the token is valid.

Raises:

HTTPException: If the token cannot be decoded or has an invalid scope.

Note:

The token is decoded using the JWT library with the provided SECRET_KEY and ALGORITHM.

get_password_hash(password: str)

Hashes the given password using the bcrypt algorithm.

Args:

password (str): The plain text password to be hashed.

Returns:

str: The hashed password as a string.

Raises:

Exception: If the password context is not initialized.

pwd_context = <CryptContext>

verify_password(plain_password, hashed_password)

Verifies if the plain password matches the hashed password.

Args:

plain_password (str): The plain text password to be verified. hashed_password (str): The hashed password to be compared with the plain text password.

Returns:

bool: Returns True if the plain password matches the hashed password, otherwise False.

Raises:

Exception: If the password context is not initialized.

REST API service Email

async src.services.email.send_email(email: EmailStr, username: str, host: str)

Get a list of contacts.

Parameters: - limit (int): The maximum number of contacts to retrieve. Default is 10, minimum is 1, and maximum is 500. - offset (int): The index of the first contact to retrieve. Default is 0. - db (AsyncSession): The database session to use for the operation. - user (User): The authenticated user making the request.

Returns: - list[ContactResponseSchema]: A list of ContactResponseSchema objects representing the contacts.

async src.services.email.send_password_reset_email(email: EmailStr, username: str, host: str)

Sends a password reset email to the specified user.

Parameters: - email (EmailStr): The email address of the user to send the password reset email to. - username (str): The username of the user to send the password reset email to. - host (str): The hostname of the website or application.

Returns: - None: This function does not return any value. It sends an email asynchronously.

Raises: - ConnectionErrors: If there is an error connecting to the email service.

Indices and tables

• Index

• Module Index

• Search Page