Welcome to SnapshotExchange’s documentation!

Is a comprehensive documentation that provides detailed information about our services and APIs. Whether you’re a developer looking to integrate with our services or an end user seeking information, you’ll find valuable resources here.

Database

class src.database.models.BGColor(value)

Bases: str, Enum

An enumeration.

black = 'black'

blue = 'blue'

brown = 'brown'

gray = 'gray'

green = 'green'

red = 'red'

transparent = 'transparent'

white = 'white'

yellow = 'yellow'

class src.database.models.BlacklistToken(**kwargs)

Bases: Base

BlacklistToken Model

This model represents a blacklisted token in the system, which is used to prevent token reuse.

Parameters:

id (int) – The unique identifier for the blacklisted token (primary key).
token (str) – The token string that has been blacklisted (unique and not nullable).
blacklisted_on (datetime) – The date and time when the token was blacklisted (default is the current time).

Example Usage:

                          blacklisted_token = BlacklistToken(
    token="your_token_string_here"
)

                        

blacklisted_on: Mapped[date]

id: Mapped[int]

token: Mapped[str]

class src.database.models.Comment(**kwargs)

Bases: Base

Comment Model

This model represents a comment left by a user on a photo in the system.

Parameters:

id (int) – The unique identifier for the comment (primary key).
text (str) – The text content of the comment (required).
created_at (datetime) – The timestamp when the comment was created (automatically generated).
updated_at (datetime) – The timestamp when the comment was last updated (automatically generated).
user_id (int) – The user identifier associated with the comment (foreign key to ‘users.id’).
photo_id (int) – The photo identifier associated with the comment (foreign key to ‘photos.id’, can be None).
update_status (bool) – A boolean flag indicating if the comment has been updated (default is False).

Example Usage:

                          comment = Comment(
    text="Your comment text here",
    user_id=1,
    photo_id=2  # Optional, can be None
)

                        

created_at: Mapped[date]

id: Mapped[int]

photo: Mapped[Photo]

photo_id: Mapped[int]

text: Mapped[int]

update_status: Mapped[bool]

updated_at: Mapped[date]

user: Mapped[int]

user_id: Mapped[int]

class src.database.models.CropMode(value)

Bases: str, Enum

An enumeration.

fill = 'fill'

fit = 'fit'

limit = 'limit'

pad = 'pad'

scale = 'scale'

thumb = 'thumb'

class src.database.models.Photo(**kwargs)

Bases: Base

Photo Model

This SQLAlchemy model represents a photo in the database.

Parameters:

id – The unique identifier for the photo (primary key).
url – The URL of the photo.
description – A brief description of the photo.
user_id – The user ID of the owner of the photo.
created_at – The timestamp when the photo was created.
cloud_public_id – The public ID of the photo in the cloud storage.
ratings – Relationship to photo ratings.
tags – Relationship to tags associated with the photo.
QR – Relationship to QR codes associated with the photo.
user – Relationship to the user who uploaded the photo.
comments – Relationship to comments on the photo.

QR: Mapped[QR_code]

cloud_public_id: Mapped[str]

comments: Mapped[list['Comment']]

created_at: Mapped[date]

description: Mapped[str]

id: Mapped[int]

ratings: Mapped[Rating]

tags: Mapped[list[str]]

url: Mapped[str]

user: Mapped[User]

user_id: Mapped[int]

class src.database.models.QR_code(**kwargs)

Bases: Base

QR_code Model

This model represents a QR code associated with a photo in the system.

Parameters:

id (int) – The unique identifier for the QR code (primary key).
url (str) – The URL or identifier of the QR code.
photo_id (int) – The ID of the associated photo (foreign key).

Example Usage:

                          new_qr_code = QR_code(
    url="https://example.com/qrcode123",
    photo_id=1
)

                        

id: Mapped[int]

photo: Mapped[int]

photo_id: Mapped[int]

url: Mapped[str]

class src.database.models.Rating(**kwargs)

Bases: Base

Rating Model

This SQLAlchemy model represents a rating given by a user to a photo in the database.

Parameters:

id – The unique identifier for the rating (primary key).
user_id – The user ID of the user who gave the rating.
rating – The numerical rating value.
photo_id – The photo ID of the photo that received the rating.
user – Relationship to the user who gave the rating.
photo – Relationship to the photo that received the rating.

id: Mapped[int]

photo: Mapped[int]

photo_id: Mapped[int]

rating: Mapped[int]

user: Mapped[User]

user_id: Mapped[int]

class src.database.models.Role(value)

Bases: Enum

User Roles Enumeration

This enumeration defines the possible roles for users in the system.

user: Represents a regular user.
moder: Represents a moderator with additional privileges.
admin: Represents an administrator with full system control.

Each role has a corresponding string value for easy identification.

Example Usage:

                          user_role = Role.user
admin_role = Role.admin

admin: str = 'Administrator'

moder: str = 'Moderator'

user: str = 'User'

class src.database.models.Tag(**kwargs)

Bases: Base

Tag Model

This model represents a tag that can be associated with photos in the system.

Parameters:

id (int) – The unique identifier for the tag (primary key).
name (str) – The name of the tag (unique and nullable).

Example Usage:

                          tag = Tag(
    name="your_tag_name_here"
)

                        

id: Mapped[int]

name: Mapped[str]

class src.database.models.User(**kwargs)

Bases: Base

User Model

SQLAlchemy model represents a user in the database.

Parameters:

id – The unique identifier for the user (primary key).
username – The username of the user (unique).
email – The email address of the user (unique).
password – The hashed password of the user.
created_at – The timestamp when the user was created.
updated_at – The timestamp when the user was last updated.
avatar – The URL of the user’s avatar image.
refresh_token – The refresh token associated with the user.
role – The role of the user (‘User’ or ‘Adminisrtator’, ‘Voderator’).
confirmed – Indicates whether the user’s email is confirmed.
is_active – Indicates whether the user’s account is active.
description – A brief description or bio of the user.
ratings – Relationship to user ratings.
photos – Relationship to user’s uploaded photos.

avatar: Mapped[str]

confirmed: Mapped[bool]

created_at: Mapped[date]

description: Mapped[str]

email: Mapped[str]

id: Mapped[int]

is_active: Mapped[bool]

password: Mapped[str]

photos: Mapped[list['Photo']]

ratings: Mapped[Rating]

refresh_token: Mapped[str]

role: Mapped[Enum]

updated_at: Mapped[date]

username: Mapped[str]

REST API routes Auth

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

Confirmation of the user’s email.

This route allows the user to confirm their email by providing the correct confirmation token.

Level of Access:

Users who have registered

Parameters:

token – str: Email confirmation token.
db – AsyncSession: Database session.

Returns:

Successful email confirmation message.

Return type:

MessageResponseSchema

Raises:

HTTPException with code 400 and detail “VERIFICATION_ERROR” in case of invalid token, and also with code 400 and detail “EMAIL_ALREADY_CONFIRMED” in case of already confirmed email.

async src.routes.auth.forgot_password(email: EmailStr, background_tasks: BackgroundTasks, request: Request, db: AsyncSession = Depends(get_db))

User password recovery request.

This route allows the user to request password recovery via email.

Level of Access:

Any user
Email is only sent to registered users.

Parameters:

email – EmailStr: Email of the user for whom password recovery is requested.
background_tasks – BackgroundTasks: Tasks that are executed in the background (sending an email with instructions).
request – Request: Client request.
db – AsyncSession: Database session.

Returns:

Message that password recovery instructions have been sent to email.

Return type:

MessageResponseSchema

Raises:

HTTPException with code 201 and detail “EMAIL_HAS_BEEN_SEND” in case of successful password recovery request.

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

Log in and get access tokens.

This route allows the user to log in by providing the correct email and password and receive access tokens.

Level of Access:

Verified users

Parameters:

body – OAuth2PasswordRequestForm: A form with the user’s email and password.
db – AsyncSession: Database Session.

Returns:

Access tokens (access_token and refresh_token).

Return type:

TokenSchema

Raises:

HTTPException with codes 401 or 403 in case of authentication or activation errors, as well as in case of invalid password.

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

Log out user and add the token to the blacklist.

This route allows the user to log out and their access token will be added to the blacklist.

Level of Access:

Current authorized user

Parameters:

credentials – HTTPAuthorizationCredentials: User authentication data (token).
db – AsyncSession: Database Session.
current_user – User: Current authenticated user.

Returns:

A message informing you that the user has successfully logged out of the system.

Return type:

MessageResponseSchema

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

Updates the user’s access token.

This route allows the user’s access token to be refreshed if a valid refresh token is present.

Level of Access:

Current authorized user

Parameters:

credentials – HTTPAuthorizationCredentials: User authentication data (access token).
db – AsyncSession: Database session.
current_user – User: Current authenticated user.

Returns:

New access tokens (access_token and refresh_token).

Return type:

TokenSchema

Raises:

HTTPException with code 401 and detail “INVALID_TOKEN” in case of invalid access token.

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

Request to confirm the user’s email.

This route allows the user to request that an email be sent to confirm the user’s email.

Level of Access:

Any user
Email is only sent to registered users.

Parameters:

body – RequestEmail: Form with user’s email.
background_tasks – BackgroundTasks: Tasks that are running in the background (sending an email).
request – Request: Client request.
db – AsyncSession: Database session.

Returns:

Email confirmation request message.

Return type:

MessageResponseSchema

Raises:

HTTPException with the code 400 and detail “EMAIL_CONFIRMED” in case of already confirmed email.

async src.routes.auth.reset_password(reset_token: str, new_password: str, db: AsyncSession = Depends(get_db))

Reset user password.

Level of Access:

Any user

This route allows the user to reset their password by providing the correct reset token and a new password.

Parameters:

reset_token – str: Password reset token.
new_password – str: The user’s new password.
db – AsyncSession: Database session.

Returns:

Password reset success message.

Return type:

MessageResponseSchema

Raises:

HTTPException with code 401 and detail “INVALID_TOKEN” in case of invalid token, and with code 401 and detail “PASWORD_RESET_SUCCESS” in case of successful password reset.

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

# Register a new user.

This route allows you to register a new user.

Level of Access:

All comers

Parameters:

body – UserSchema: New user data.
background_tasks – BackgroundTasks: Tasks to be performed in the background.
request – Request: Inquiry.
session – AsyncSession: Database Session.

Returns:

New user data and a message about successful registration.

Raises:

HTTPException with code 409 and detail “ALREADY_EXISTS_EMAIL” or “ALREADY_EXISTS_USERNAME” if a user with that email or username already exists.

REST API routes Users

async src.routes.users.activate_user(email: EmailStr, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Activate a user by email.

This route allows to activate a user by their email.

Level of Access:

Administartor

Parameters:

email – EmailStr: Email of the user to activate.
db – AsyncSession: Database session.

Returns:

Successful user blocking message or error message.

Return type:

dict

async src.routes.users.assign_role(email: EmailStr, role: Role, db: AsyncSession = Depends(get_db), redis_client: Redis = Depends(init_async_redis))

Assign a role to a user by email.

This route allows to assign the selected role to a user by their email.

Level of Access:

Administartor

Parameters:

email – EmailStr: Email of the user to whom you want to assign the role.
selected_role – Role: The selected role for the assignment (Administrator, Moderator or User).
db – AsyncSession: Database Session.
redis_client – Redis: Redis client.

Returns:

Message about successful role change.

Return type:

dict

async src.routes.users.ban_user(email: EmailStr, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Block a user by email.

This route allows to block a user by their email.

Level of Access:

Administartor

Parameters:

email – EmailStr: Email of the user to block.
db – AsyncSession: Database session.

Returns:

Successful user blocking message or error message.

Return type:

dict

async src.routes.users.edit_my_profile(avatar: UploadFile = File(PydanticUndefined), new_username: str = Form(None), new_description: str = Form(None), current_user: User = Depends(get_authenticated_user), redis_client: Redis = Depends(init_async_redis), db: AsyncSession = Depends(get_db))

Edit the current user’s profile.

This route allows the current user to edit own profile, including uploading an avatar, changing the username and description.

Level of Access: - Current authorized user

Parameters:

avatar – UploadFile: User avatar file (optional).
new_username – str: New username (optional).
new_description – str: New user description (optional).
current_user – User: Current authenticated user.
redis_client – Redis: Redis client.
db – AsyncSession: Database session.

Returns:

Updated user profile.

Return type:

UserDb

Raises:

HTTPException with code 400 and detail “USER_EXISTS” if the new username already exists.

async src.routes.users.get_users(skip: int = 0, limit: int = 10, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Get a list of users.

This route allows to get a list of pagination-aware users.

Level of Access:

Current authorized user

Parameters:

skip – int: Number of users to skip.
limit – int: Maximum number of users to return.
current_user – User: Current authenticated user.
db – AsyncSession: Database session.

Returns:

List of users.

Return type:

List[UserDb]

async src.routes.users.read_my_profile(current_user: User = Depends(get_authenticated_user))

Get the profile of the current user.

This route retrieves the profile of the current authenticated user.

Level of Access:

Current authorized user

Parameters:: current_user – User: The current authenticated user.
Returns:: Current user profile.
Return type:: UserDb

async src.routes.users.user_profile(username: str, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db)) → dict | None

Get a user’s profile by username.

This route allows to retrieve a user’s profile by their username.

Level of Access:

Current authorized user

Parameters:

username – str: The username of the user whose profile is to be retrieved.
current_user – User: The current authenticated user (optional).
db – AsyncSession: Database session.

Returns:

User profile or None if no user is found.

Return type:

dict | None

Raises:

HTTPException with code 404 and detail “NOT_FOUND” if the user is not found.

REST API routes Photos

async src.routes.photos.get_all_photos(skip: int = 0, limit: int = 10, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db)) → list

Get All Photos

This endpoint retrieves a list of photos from the database.

Parameters:

skip (int) – The number of photos to skip (default is 0).
limit (int) – The maximum number of photos to retrieve (default is 10).
current_user (User) – The authenticated user (Dependency).
db (AsyncSession) – The asynchronous database session (Dependency).

Returns:

A list of Photo objects.

Return type:

list[PhotosDb]

Raises:

HTTPException 401 – Unauthorized if the user is not authenticated.
HTTPException 500 – Internal Server Error if there’s a database issue.

Example Request:

                          GET /get_all?skip=0&limit=10 HTTP/1.1
Host: yourapi.com
Authorization: Bearer your_access_token

                        

Example Response:

                          HTTP/1.1 200 OK
Content-Type: application/json

                          [
{
    "id": 1,
    "title": "Photo 1",
    "url": "https://example.com/photo1.jpg"
},
{
    "id": 2,
    "title": "Photo 2",
    "url": "https://example.com/photo2.jpg"
}
]

                        

This documentation provides information about the /get_all endpoint, its parameters, the expected response, and potential error responses. You can include this in your Sphinx documentation for your API. If you need further details or have any specific requirements, please let me know.

async src.routes.photos.get_my_photos(skip: int = 0, limit: int = 10, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db)) → list

Get My Photos

This endpoint retrieves a list of photos that belong to the authenticated user from the database.

Parameters:

skip (int) – The number of photos to skip (default is 0).
limit (int) – The maximum number of photos to retrieve (default is 10).
current_user (User) – The authenticated user (Dependency).
db (AsyncSession) – The asynchronous database session (Dependency).

Returns:

A list of Photo objects.

Return type:

list[PhotosDb]

Raises:

HTTPException 401 – Unauthorized if the user is not authenticated.
HTTPException 500 – Internal Server Error if there’s a database issue.

Example Request:

                          GET /get_my?skip=0&limit=10 HTTP/1.1
Host: yourapi.com
Authorization: Bearer your_access_token

                        

Example Response:

                          HTTP/1.1 200 OK
Content-Type: application/json

                          [
{
    "id": 1,
    "title": "My Photo 1",
    "url": "https://example.com/my_photo1.jpg"
},
{
    "id": 2,
    "title": "My Photo 2",
    "url": "https://example.com/my_photo2.jpg"
}
]

                        

async src.routes.photos.get_one_photo(photo_id: int, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Get One Photo by ID

This endpoint retrieves a specific photo by its unique ID.

Parameters:

photo_id (int) – The ID of the photo to retrieve.
current_user (User) – The authenticated user (Dependency).
db (AsyncSession) – The asynchronous database session (Dependency).

Returns:

The Photo object with the specified ID.

Return type:

PhotosDb

Raises:

HTTPException 204 – No Content if no photo with the given ID is found.
HTTPException 401 – Unauthorized if the user is not authenticated.
HTTPException 429 – Too Many Requests if the rate limit is exceeded.
HTTPException 500 – Internal Server Error if there’s a database issue.

Example Request:

                          GET /123 HTTP/1.1
Host: yourapi.com
Authorization: Bearer your_access_token

                        

Example Response:

The response contains the Photo object with the specified ID.

                          HTTP/1.1 200 OK
Content-Type: application/json

                          {
    "id": 123,
    "title": "Photo 123",
    "url": "https://example.com/photo123.jpg"
}

                        

Rate Limiting:

This endpoint is rate-limited to no more than 10 requests per minute.

async src.routes.photos.make_URL_QR(photo_id: int, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Make QR Code for Photo URL

This endpoint generates a QR code for a specific photo URL.

Parameters:

photo_id (int) – The ID of the photo for which the QR code is generated.
current_user (User) – The authenticated user (Dependency).
db (AsyncSession) – The asynchronous database session (Dependency).

Returns:

The generated QR code data.

Return type:

bytes

Raises:

HTTPException 401 – Unauthorized if the user is not authenticated.
HTTPException 429 – Too Many Requests if the rate limit is exceeded.
HTTPException 500 – Internal Server Error if there’s a database issue.

Example Request:

                          POST /make_QR/?photo_id=123 HTTP/1.1
Host: yourapi.com
Authorization: Bearer your_access_token

                        

Example Response:

The response contains the generated QR code data, which can be used to display or download the QR code image.

                          HTTP/1.1 200 OK
Content-Type: application/octet-stream

[QR code binary data]

Rate Limiting:

This endpoint is rate-limited to no more than 10 requests per minute.

async src.routes.photos.patch_pdate_photo(photo_id: int, new_photo_description: str, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Update Photo Description

This endpoint updates the description of a specific photo by its unique ID.

Parameters:

photo_id (int) – The ID of the photo to update.
new_photo_description (str) – The new description for the photo.
current_user (User) – The authenticated user (Dependency).
db (AsyncSession) – The asynchronous database session (Dependency).

Returns:

The updated Photo object with the specified ID.

Return type:

PhotosDb

Raises:

HTTPException 404 – Not Found if no photo with the given ID is found.
HTTPException 401 – Unauthorized if the user is not authenticated.
HTTPException 429 – Too Many Requests if the rate limit is exceeded.
HTTPException 500 – Internal Server Error if there’s a database issue.

Example Request:

                          PATCH /123 HTTP/1.1
Host: yourapi.com
Authorization: Bearer your_access_token
Content-Type: application/json

{
    "new_photo_description": "Updated description for photo 123"
}

                        

Example Response:

The response contains the updated Photo object with the specified ID.

                          HTTP/1.1 200 OK
Content-Type: application/json

                          {
    "id": 123,
    "title": "Photo 123",
    "url": "https://example.com/photo123.jpg",
    "description": "Updated description for photo 123"
}

                        

Rate Limiting:

This endpoint is rate-limited to no more than 10 requests per minute.

async src.routes.photos.remove_photo(photo_id: int, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db)) → MessageResponseSchema

Remove Photo by ID

This endpoint removes a specific photo by its unique ID.

Parameters:

photo_id (int) – The ID of the photo to remove.
current_user (User) – The authenticated user (Dependency).
db (AsyncSession) – The asynchronous database session (Dependency).

Returns:

A message response indicating the success of the removal.

Return type:

MessageResponseSchema

Raises:

HTTPException 404 – Not Found if no photo with the given ID is found.
HTTPException 401 – Unauthorized if the user is not authenticated.
HTTPException 500 – Internal Server Error if there’s a database issue.

Example Request:

                          DELETE /123 HTTP/1.1
Host: yourapi.com
Authorization: Bearer your_access_token

                        

Example Response:

The response contains a message indicating that the photo has been successfully removed.

                          HTTP/1.1 200 OK
Content-Type: application/json

                          {
    "message": "Photo removed successfully"
}

                        

async src.routes.photos.upload_photo(photo_file: UploadFile = File(PydanticUndefined), description: str | None = Form(None), tags: list[str] = Form(None), width: int | None = Form(None), height: int | None = Form(None), crop_mode: CropMode = Form(None), rounding: int | None = Form(None), background_color: BGColor = Form(None), rotation_angle: int | None = Form(None), current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db)) → PhotosDb

Upload a Photo

This endpoint allows users to upload a new photo with various customization options.

Parameters:

photo_file (UploadFile) – The photo file to upload (required).
description (str | None) – An optional description for the photo (string, max 500 characters).
tags (list[str]) – Tags to associate with the photo (list of strings, max 5 tags, each tag max 25 characters).
width (int | None) – The desired width for the photo transformation (integer).
height (int | None) – The desired height for the photo transformation (integer).
crop_mode (CropMode | None) – The cropping mode for the photo transformation (string).
rounding (int | None) – Rounding photo corners (in pixels).
background_color (BGColor | None) – The background color for the photo transformation (string).
rotation_angle (int | None) – The angle for the photo transformation (integer).
current_user (User) – The authenticated user making the upload request.
db (AsyncSession) – The asynchronous database session (Dependency).

Returns:

The uploaded photo information.

Return type:

PhotosDb

Raises:

HTTPException 400 – If the description is too long or if there are too many tags.
HTTPException 400 – If any tag name is too long.
HTTPException 400 – If the cropping mode or background color is invalid.

Example Response:

An object containing the uploaded photo information.

REST API routes Comments

async src.routes.comments.change_comment(comment_id: int = Form(PydanticUndefined), text: str = Form(PydanticUndefined), current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Update a comment.

Parameters:

comment_id (int) – ID of the comment to be updated.
text (str) – The new text for the comment.
current_user (User) – The authenticated user.
db (AsyncSession) – Database session.

Returns:

Updated comment if successful.

Return type:

dict

Raises:

HTTPException 404 – If the comment does not exist.
HTTPException 403 – If the current user is not the author of the comment.

async src.routes.comments.post_comment(photo_id: int = Form(PydanticUndefined), text: str = Form(PydanticUndefined), current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Publish a comment on a photo.

Parameters:

photo_id (int) – ID of the photo to comment on.
text (str) – The text of the comment.
current_user (User) – The authenticated user.
db (AsyncSession) – Database session.

Returns:

Comment object if successful.

Return type:

dict

Raises:

HTTPException – If the comment could not be created.

Return type:

HTTPException

async src.routes.comments.remove_comment(comment_id: int = Form(PydanticUndefined), current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Delete a comment.

Parameters:

comment_id (int) – ID of the comment to be deleted.
current_user (User) – The authenticated user.
db (AsyncSession) – Database session.

Returns:

A response indicating the success of the deletion.

Return type:

dict

Raises:

HTTPException 404 – If the comment does not exist.
HTTPException 403 – If the current user is not the author of the comment and is not a moderator or admin.

async src.routes.comments.show_photo_comments(photo_id: int, limit: int = 0, offset: int = 10, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Retrieve comments for a specific photo.

Parameters:

photo_id (int) – The ID of the photo for which comments are to be retrieved.
limit (int) – The maximum number of comments to retrieve (default is 0, which retrieves all).
offset (int) – The offset for paginating through comments (default is 10).
current_user (User) – The authenticated user.
db (AsyncSession) – Database session.

Returns:

A dictionary containing the list of comments for the specified photo.

Return type:

dict

Raises:

HTTPException 404 – If the photo does not exist.

async src.routes.comments.show_user_comments(user_id: int, limit: int = 0, offset: int = 10, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Retrieve comments for a specific user.

Parameters:

user_id (int) – The ID of the user for whom comments are to be retrieved.
limit (int) – The maximum number of comments to retrieve (default is 0, which retrieves all).
offset (int) – The offset for paginating through comments (default is 10).
current_user (User) – The authenticated user.
db (AsyncSession) – Database session.

Returns:

A dictionary containing the list of comments for the specified user.

Return type:

dict

Raises:

HTTPException 404 – If the user does not exist.

REST API routes Ratings

async src.routes.ratings.created_rating(photo_id: int, rating: int, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Create a new rating for a photo.

This function allows users to create a new rating for a photo. Users must be authenticated. The function accepts the rating value, the ID of the photo being rated, the current authenticated user, and a database session.

Parameters:

rating – int: The rating value.
photo_id – int: The ID of the photo being rated.
current_user – User: The currently authenticated user.
db – AsyncSession: The database session.

Returns:

The newly created rating record.

Return type:

RatingSchema

async src.routes.ratings.delete_rating_ADmin_Moder(photo_id: int, user_id: int, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Delete all ratings for a photo (Admin/Moderator only).

This function deletes all ratings for a photo with the specified ID and user ID from the database. Only users with the ‘admin’ or ‘moder’ role can access this endpoint.

Parameters:

photo_id – int: The ID of the photo.
user_id – int: The ID of the user.
current_user – User: The currently authenticated user (admin or moder).
db – AsyncSession: The database session.

Returns:

A message indicating successful deletion.

Return type:

MessageResponseSchema

async src.routes.ratings.get_rating(photo_id: int, db: AsyncSession = Depends(get_db))

Get the average rating for a photo.

This function retrieves the average rating for a photo with the specified ID from the database.

Parameters:

photo_id – int: The ID of the photo.
db – AsyncSession: The database session.

Returns:

The average rating for the photo.

Return type:

float

async src.routes.ratings.get_rating_Admin_Moder(photo_id: int, current_user: User = Depends(get_authenticated_user), db: AsyncSession = Depends(get_db))

Get all ratings for a photo (Admin/Moderator only).

This function retrieves all ratings for a photo with the specified ID from the database. Only users with the ‘admin’ or ‘moder’ role can access this endpoint.

Parameters:

photo_id – int: The ID of the photo.
current_user – User: The currently authenticated user (admin or moder).
db – AsyncSession: The database session.

Returns:

All ratings for the photo.

Return type:

List[Rating]

REST API routes Search

async src.routes.search.admin_search(user_id: int, text: str = Query(None), rating_low: float = Query(0), rating_high: float = Query(999), start_data: str = Query(1963-09-24), end_data: str = Query(2023-09-09), current_user: ~src.database.models.User = Depends(get_authenticated_user), db: ~sqlalchemy.ext.asyncio.session.AsyncSession = Depends(get_db))

Search photos for administrators by “user_id” with optional filtering by “tag” or “description”, “rating”, and “date range”.

This function allows administrators to search for photos based on a provided “user_id” and filter search results by matching “text” with “tags”(if text start with “#”) or matching with “descriptions”, minimum and maximum “rating values”, as well as a “date range”. Parameter “text” must be a string minimum 2 characters. Parameter “rating” is float number. Parameter “date range” is a string in the format “YYYY-MM-DD” The search is performed within the “user_id” parameter and optional “text”, “rating range” and “date range”

Parameters:

text – str: The search text or tag to filter photos.
user_id – int: The user ID to filter photos. Leave as None to search all users’ photos.
rating_low – float: The minimum rating value to filter photos.
rating_high – float: The maximum rating value to filter photos.
start_data – str: The start date for the search range in ‘YYYY-MM-DD’ format.
end_data – str: The end date for the search range in ‘YYYY-MM-DD’ format.
current_user – User: The currently authenticated user (administrator).
db – AsyncSession: The database session.

Returns:

A list of photos that match the search criteria.

Return type:

List[Photo]

async src.routes.search.search_description(description: str = Path(PydanticUndefined), rating_low: float = Query(0), rating_high: float = Query(999), start_data: str = Query(1963-09-24), end_data: str = Query(2023-09-09), current_user: ~src.database.models.User = Depends(get_authenticated_user), db: ~sqlalchemy.ext.asyncio.session.AsyncSession = Depends(get_db))

Search photos with matching in “description” and optional filtering by “rating”, and “date range”.

This function allows users to search for photos based on a provided text. Users can also filter search results by specifying minimum and maximum “rating values”, as well as a “date range”. Parameter “text” must be a string minimum 2 characters. Parameter “rating” is float number. Parameter “date range” is a string in the format “YYYY-MM-DD” The search is performed within the specified parameters.

Parameters:

description – str: The search text or tag to filter photos.
rating_low – float: The minimum rating value to filter photos.
rating_high – float: The maximum rating value to filter photos.
start_data – str: The start date for the search range in ‘YYYY-MM-DD’ format.
end_data – str: The end date for the search range in ‘YYYY-MM-DD’ format.
current_user – User: The currently authenticated user.
db – AsyncSession: The database session.

Returns:

A list of photos that match the search criteria.

Return type:

List[Photo]

async src.routes.search.search_tag(tag: str = Path(PydanticUndefined), rating_low: float = Query(0), rating_high: float = Query(999), start_data: str = Query(1963-09-24), end_data: str = Query(2023-09-09), current_user: ~src.database.models.User = Depends(get_authenticated_user), db: ~sqlalchemy.ext.asyncio.session.AsyncSession = Depends(get_db))

Search photos with “tag” optional filtering by “rating”, and “date range”.

This function allows users to search for photos based on a provided “tag”. Users can also filter search results by specifying minimum and maximum “rating values”, as well as a “date range”. Parameter “tag” must be a string minimum 2 characters. Parameter “rating” is float number. Parameter “date range” is a string in the format “YYYY-MM-DD” The search is performed within the specified parameters.

Parameters:

tag – str: The search text or tag to filter photos.
rating_low – float: The minimum rating value to filter photos.
rating_high – float: The maximum rating value to filter photos.
start_data – str: The start date for the search range in ‘YYYY-MM-DD’ format.
end_data – str: The end date for the search range in ‘YYYY-MM-DD’ format.
current_user – User: The currently authenticated user.
db – AsyncSession: The database session.

Returns:

A list of photos that match the search criteria.

Return type:

List[Photo]

VIEWS routes Dashboard

async src.views.dashboard.root(request: Request)

Project Information Page

This endpoint serves an HTML page displaying information about the project.

Parameters:: request (Request) – The HTTP request object.
Returns:: The HTML page displaying project information.
Return type:: HTMLResponse

async src.views.dashboard.show_route_list(request: Request)

VIEWS routes Photos

async src.views.photos.search_by_tag(request: ~starlette.requests.Request, query: str, search_type: str, skip: int = 0, limit: int = 10, access_token: str = Cookie(None), rating_low: float = Query(0), rating_high: float = Query(999), start_data: str = Query(1963-09-24), end_data: str = Query(2023-09-09), db: ~sqlalchemy.ext.asyncio.session.AsyncSession = Depends(get_db))

async src.views.photos.view_database(request: Request, skip: int = 0, limit: int = 10, access_token: str = Cookie(None), db: AsyncSession = Depends(get_db))

View All Photos

This endpoint retrieves and displays a view of all photos from the database.

Parameters:

request (Request) – The HTTP request object.
skip – The number of photos to skip (default is 0).
limit – The maximum number of photos to retrieve (default is 10).
current_user (User) – The authenticated user (Dependency).
db (AsyncSession) – The asynchronous database session (Dependency).

Returns:

A rendered HTML page displaying photos with additional information.

Return type:

templates.TemplateResponse

Raises:

HTTPException 500 – Internal Server Error if there’s a database issue.

Example Request:

                          GET /view_all_photos?skip=0&limit=10 HTTP/1.1
Host: yourapi.com

Example Response:

A rendered HTML page displaying a list of photos with usernames, descriptions, comments, tags, and creation timestamps.

VIEWS routes Users

async src.views.users.view_user_profile(username: str, request: Request, access_token: str = Cookie(None), db: AsyncSession = Depends(get_db))

View User Profile

This endpoint retrieves and displays a user’s profile based on their username.

Parameters:

username (str) – The username of the user whose profile to view.
request (Request) – The HTTP request object.
db (AsyncSession) – The asynchronous database session (Dependency).

Returns:

The HTML page displaying the user’s profile.

Return type:

templates.TemplateResponse

Raises:

HTTPException 404 – User not found if the user with the specified username is not found.

VIEWS routes Chat

async src.views.chat.chat(request: Request)

Get Chat HTML Page

This endpoint serves an HTML page for a chat application.

Parameters:: request (Request) – The HTTP request object.
Returns:: The HTML page for the chat.
Return type:: HTMLResponse

async src.views.chat.websocket_endpoint(websocket: WebSocket)

WebSocket Endpoint

This WebSocket endpoint establishes a connection for real-time chat. Clients can send and receive text messages.

Parameters:: websocket (WebSocket) – WebSocket connection.

Repository Users

async src.repository.users.activate_user(email: str, db: AsyncSession) → None

Activates a user account by setting their ‘is_active’ status to True.

Parameters:

email (str) – The email address of the user to activate.
db (AsyncSession) – The asynchronous database session.

Returns:

None

Return type:

None

async src.repository.users.add_to_blacklist(token: str, db: AsyncSession) → None

Adds a token to the blacklist.

Parameters:

token – str: Pass the token to be blacklisted
db – AsyncSession: Create a new session with the database

Returns:

None

async src.repository.users.ban_user(email: str, db: AsyncSession) → None

The ban_user function takes in an email and a database session. It then finds the user with that email, sets their is_active field to False, and commits the change to the database.

Parameters:

email – str: Identify the user to be banned
db – Session: Pass in the database session

Returns:

None, because we don’t need to return anything

async src.repository.users.confirm_email(email: str, db: AsyncSession) → None

The confirmed_email function sets the confirmed field of a user to True.

Parameters:

email – str: Get the email of the user that is trying to confirm their account
db – Session: Pass the database session to the function

Returns:

None

async src.repository.users.create_user(body: UserSchema, db: AsyncSession) → User

Create a new user in the database.

Parameters:

body (UserSchema) – The user data.
db (AsyncSession) – The database session.

Returns:

The created user object.

Return type:

User

async src.repository.users.edit_my_profile(file, new_description, new_username, user: User, db: AsyncSession) → User

The edit_my_profile function allows a user to edit their profile.

Parameters:

file – Upload the image to cloudinary
new_username – Change the username of the user
user – User: Get the user object from the database
db – AsyncSession: Access the database

Returns:

A user object

async src.repository.users.get_user_by_email(email: str, db: AsyncSession) → User

The get_user_by_email function takes in an email and a database session, then returns the user with that email.

Parameters:

email – str: Get the email from the user
db – Session: Pass a database session to the function

Returns:

A user object if the email is found in the database

async src.repository.users.get_user_by_reset_token(reset_token: str, db: AsyncSession) → User | None

Retrieve a user by their reset token.

This function queries the database to retrieve a user based on their reset token.

Parameters:

reset_token – str: The reset token associated with the user.
db – AsyncSession: The asynchronous database session.

Returns:

The user if found, or None if not found.

Return type:

User | None

async src.repository.users.get_user_by_user_id(user_id: int, db: AsyncSession) → User | None

Get User by User ID

This function retrieves a user by their user ID from the database.

Parameters:

user_id (int) – The ID of the user to retrieve.
db (AsyncSession) – An asynchronous database session.

Returns:

The user with the specified ID, or None if the user is not found.

Return type:

User | None

async src.repository.users.get_user_by_username(username: str, db: AsyncSession) → User | None

The get_user_by_email function takes in an email and a database session, then returns the user with that email.

Parameters:

email – str: Get the email from the user
db – Session: Pass a database session to the function

Returns:

A user object if the email is found in the database

async src.repository.users.get_user_profile(username: str, db: AsyncSession) → User

Get the profile of a user by username.

Parameters:

username (str) – The username of the user.
db (AsyncSession) – The database session.

Returns:

The user profile.

Return type:

User

async src.repository.users.get_users(skip: int, limit: int, db: AsyncSession) → list[src.database.models.User]

The get_users function returns a list of all users from the database.

Parameters:

skip – int: Skip the first n records in the database
limit – int: Limit the number of results returned
db – AsyncSession: Pass the database session to the function

Returns:

A list of all users

async src.repository.users.is_blacklisted_token(token: str, db: AsyncSession) → bool

Check if a Token is Blacklisted

This function checks if a given token is blacklisted in the database.

Parameters:

token (str) – The token to be checked.
db (AsyncSession) – An asynchronous database session.

Returns:

True if the token is blacklisted, False otherwise.

Return type:

bool

async src.repository.users.make_user_role(email: str, role: Role, db: AsyncSession) → None

The make_user_role function takes in an email and a role, and then updates the user’s role to that new one. Args: email (str): The user’s email address. role (Role): The new Role for the user.

Parameters:

email – str: Get the user by email
role – Role: Set the role of the user
db – Session: Pass the database session to the function

Returns:

None

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

The update_token function updates the refresh token for a user.

Parameters:

user – User: Identify the user that is being updated
token – str | None: Store the token in the database
db – Session: Create a database session

Returns:

None, but the return type is specified as str | none

Repository Photos

async src.repository.photos.get_URL_QR(photo_id: int, db: AsyncSession)

Generate and retrieve a QR code URL for a photo.

Parameters:

photo_id (int) – The ID of the photo for which to generate a QR code.
db (AsyncSession) – The database session.

Returns:

A dictionary containing the source URL and QR code URL.

Return type:

dict

async src.repository.photos.get_my_photos(skip: int, limit: int, current_user: User, db: AsyncSession) → list[src.database.models.Photo]

The get_photos function returns a list of all photos of current_user from the database.

Parameters:

skip – int: Skip the first n records in the database
limit – int: Limit the number of results returned
current_user – User
db – AsyncSession: Pass the database session to the function

Returns:

A list of all photos

async src.repository.photos.get_or_create_tag(tag_name: str, db: AsyncSession) → Tag

Get or Create Tag

This function retrieves an existing tag with the specified name from the database or creates a new one if it doesn’t exist.

Parameters:

tag_name (str) – The name of the tag to retrieve or create.
db (AsyncSession) – The asynchronous database session.

Returns:

The existing or newly created Tag object.

Return type:

Tag

Raises:

Exception – Raises an exception if there’s an issue with database operations.

Example Usage:

                          tag_name = "example_tag"
async with get_db() as db:
    tag = await get_or_create_tag(tag_name, db)
    print(tag.name)  # Print the name of the existing or newly created tag.

                        

This function first attempts to retrieve an existing tag with the specified name from the database. If the tag exists, it returns that tag. If not, it creates a new tag with the provided name, adds it to the database, commits the transaction, and returns the newly created tag.

async src.repository.photos.get_photo_by_id(photo_id: int, db: AsyncSession) → dict

Retrieve a photo by its ID from the database.

Parameters:

photo_id – int: The ID of the photo to retrieve
db – AsyncSession: Pass the database session to the function

Returns:

A dictionary containing the details of the retrieved photo, or None if not found

async src.repository.photos.get_photo_comments(photo_id: int, db: AsyncSession) → list[str]

async src.repository.photos.get_photo_info(photo: Photo, db: AsyncSession)

async src.repository.photos.get_photo_tags(photo_id: int, db: AsyncSession) → list[str] | None

Get Photo Tags

This function retrieves a list of tags associated with a specific photo by its ID.

Parameters:

photo_id (int) – The ID of the photo for which to retrieve tags.
db (AsyncSession) – The asynchronous database session.

Returns:

A list of tag names associated with the photo, or None if there are no tags.

Return type:

list[str] | None

Raises:

Exception – Raises an exception if there’s an issue with database operations.

Example Usage:

                          photo_id = 123
async with get_db() as db:
    tags = await get_photo_tags(photo_id, db)
    if tags:
        print(tags)  # Print the list of tag names associated with the photo.
    else:
        print("No tags found for the photo.")

                        

This function constructs a database query to retrieve tag names associated with a specific photo. If tags are found, it returns a list of tag names. If no tags are associated with the photo, it returns None.

async src.repository.photos.get_photos(skip: int, limit: int, db: AsyncSession) → list[src.database.models.Photo]

The get_photos function returns a list of all photos of current_user from the database.

Parameters:

skip – int: Skip the first n records in the database
limit – int: Limit the number of results returned
current_user – User
db – AsyncSession: Pass the database session to the function

Returns:

A list of all photos

async src.repository.photos.remove_photo(photo_id: int, user: User, db: AsyncSession) → bool

Remove a photo from cloud storage and the database.

Parameters:

photo_id (int) – The ID of the photo to remove.
db (AsyncSession) – The database session.
user (User) – The user who is removing the photo.

Returns:

True if the removal was successful, False otherwise.

Return type:

bool

async src.repository.photos.update_photo(current_user: User, photo_id: int, description: str, db: AsyncSession) → dict

Update the description of a photo owned by the current user.

Parameters:

current_user – User: The user who owns the photo
photo_id – int: The ID of the photo to be updated
description – str: The new description for the photo
db – AsyncSession: Pass the database session to the function

Returns:

A dictionary containing the updated photo’s URL and description

async src.repository.photos.upload_photo(current_user: User, photo: File(PydanticUndefined), description: str | None, db: AsyncSession, width: int | None, height: int | None, crop_mode: str | None, rounding, background_color, rotation_angle: int | None, tags: List[str] = []) → Photo

Upload a Photo

This function allows users to upload a new photo with various customization options.

Parameters:

current_user (User) – The authenticated user uploading the photo.
photo (File) – The photo file to upload.
description (str | None) – An optional description for the photo (string, max 500 characters).
db (AsyncSession) – The asynchronous database session.
width (int | None) – The desired width for the photo transformation (integer).
height (int | None) – The desired height for the photo transformation (integer).
crop_mode (str | None) – The cropping mode for the photo transformation (string).
rounding – Rounding photo corners (in pixels).
background_color – The background color for the photo transformation (string).
rotation_angle (int | None) – The angle for the photo transformation (integer).
tags (List[str]) – Tags to associate with the photo (list of strings).

Returns:

The uploaded photo.

Return type:

Photo

Raises:

HTTPException 400 – If the description is too long.
HTTPException 400 – If any tag name is too long.
HTTPException 400 – If the cropping mode or background color is invalid.

Example Request:

                          uploaded_photo = await upload_photo(
    current_user,
    photo,
    "A beautiful landscape",
    db,
    800,
    600,
    "crop",
    10,
    "white",
    90,
    ["nature", "scenic"]
)

                        

Example Response:

An object containing the uploaded photo information.

Repository Comments

async src.repository.comments.create_comment(content: str, user: str, photos_id: int, db: AsyncSession)

Creates a new comment and stores it in the database.

Parameters:

content – str: Text of the comment.
user – str: The user who left the comment.
photos_id – int: The ID of the photo to which the comment is linked.
db – AsyncSession: Database session to perform operations.

Returns:

Comment: Comment created.

Raises:

Exception – If an error occurred while creating the comment.

async src.repository.comments.delete_comment(id: int, db: AsyncSession)

Delete a comment

Parameters:

id – The ID of the comment to update
db (AsyncSession) – The database session.

Returns:

The comment object.

Return type:

Comment

async src.repository.comments.get_comment(id: int, db: AsyncSession)

Get a Comment by ID

This function retrieves a comment by its ID from the database.

Parameters:

id (int) – The ID of the comment to retrieve.
db (AsyncSession) – An asynchronous database session.

Returns:

The comment with the specified ID, or None if the comment is not found.

Return type:

Comment | None

async src.repository.comments.get_photo_comments(offset: int, limit: int, photo_id: int, db: AsyncSession)

Gets comments on a specific photo with pagination.

Parameters:

offset – int: Offset to sample comments.
limit – int: Maximum number of comments to sample.
photo_id – int: Identifier of the photo to which the comments refer.
db – AsyncSession: The database session for performing operations.

Returns:

list[Comment]: Pagination-aware list of comments on the photo.

async src.repository.comments.get_user_comments(offset: int, limit: int, user_id: int, db: AsyncSession)

Review comments by photo

Parameters:

limit – limit of comments
offset (int) – offset of comments
photos_id (int) – The ID of the photo for which to retrieve ratings.
db (AsyncSession) – The database session.

Type:

int

Returns:

A list of comment objects.

Return type:

list[Comment]

async src.repository.comments.update_comment(text: str, id: int, db: AsyncSession)

Update a comment

Parameters:

text – The text for updating the comment
id (int) – The ID of the comment to update
db (AsyncSession) – The database session.

Returns:

The updated comment object.

Return type:

Comment

Repository Ratings

async src.repository.ratings.create_rating(rating: int, photos_id: int, user: User, db: AsyncSession)

Create a new rating for a photo in the database.

Parameters:

rating (str) – The rating value.
photos_id (str) – The ID of the photo to rate.
user (User) – The user who is creating the rating.
db (AsyncSession) – The database session.

Returns:

The created rating object.

Return type:

Rating

async src.repository.ratings.delete_all_ratings(photos_id: int, user_id: int, db: AsyncSession)

Delete all ratings by a specific user for a photo.

Parameters:

photos_id (str) – The ID of the photo for which to delete ratings.
user_id (str) – The ID of the user whose ratings should be deleted.
db (AsyncSession) – The database session.

Returns:

A message indicating whether ratings were deleted.

Return type:

dict

async src.repository.ratings.get_all_ratings(photos_id: int, db: AsyncSession)

Retrieve all ratings for a photo.

Parameters:

photos_id (str) – The ID of the photo for which to retrieve ratings.
db (AsyncSession) – The database session.

Returns:

A list of rating objects.

Return type:

List[Rating]

async src.repository.ratings.get_rating(photos_id: int, db: AsyncSession)

Calculate and retrieve the average rating for a photo.

Parameters:

photos_id (str) – The ID of the photo for which to calculate the rating.
db (AsyncSession) – The database session.

Returns:

The average rating for the photo.

Return type:

float

Repository Search

async src.repository.search.search_admin(user_id: int, text: str, rating_low: float, rating_high: float, start_data: date, end_data: date, db: AsyncSession) → [<class 'src.database.models.Photo'>]

Search photos for admin by user with optional filtering by tag, rating and date range.

Parameters:

text (str) – Search text
user_id (int) – user id for search
rating_low (float) – Minimum rating
rating_high (float) – Maximum rating
start_data (Date) – Start date for the search range.
end_data (Date) – End date for the search range.
db (AsyncSession) – The database session

Returns:

List of photos

Return type:

List[Photo]

async src.repository.search.search_by_description(text: str, rating_low: float, rating_high: float, start_data: date, end_data: date, db: AsyncSession) → [<class 'src.database.models.Photo'>]

Search photos by match in description, with optional filtering by rating and date range.

Parameters:

text (str) – Search text
rating_low (float) – Minimum rating
rating_high (float) – Maximum rating
start_data (Date) – Start date for the search range.
end_data (Date) – End date for the search range.
db (AsyncSession) – The database session

Returns:

List of photos

Return type:

List[Photo]

async src.repository.search.search_by_tag(tag: str, rating_low: float, rating_high: float, start_data: date, end_data: date, db: AsyncSession) → [<class 'src.database.models.Photo'>]

Search photos by match in tags, with optional filtering by rating and date range.

Parameters:

tag (str) – Search text
rating_low (float) – Minimum rating
rating_high (float) – Maximum rating
start_data (Date) – Start date for the search range.
end_data (Date) – End date for the search range.
db (AsyncSession) – The database session

Returns:

List of photos

Return type:

List[Photo]

async src.repository.search.search_by_username(username: str, db: AsyncSession) → list[src.database.models.Photo]

Search for a user by their username and retrieve a list of their photos.

Parameters:

username (str) – The username to search for.
db (AsyncSession) – The asynchronous database session.

Returns:

A list of photos uploaded by the user.

Return type:

list[Photo]

Service Auth

class src.services.auth.Auth

Bases: object

The Auth class provides authentication-related functionality for the application.

It includes methods for password hashing, token creation and decoding, and user verification.

Attributes:: pwd_context (CryptContext): An instance of CryptContext for password hashing. SECRET_KEY (str): The secret key used for token generation and decoding. ALGORITHM (str): The algorithm used for token encoding and decoding. oauth2_scheme (OAuth2PasswordBearer): An instance of OAuth2PasswordBearer for OAuth2 authentication.
Methods:: verify_password(plain_password, hashed_password): Verify a plain password against a hashed password. get_password_hash(password): Generate a hashed password from a plain text password. create_access_token(data, expires_delta=None): Create an access token with an optional expiration time. create_refresh_token(data, expires_delta=None): Create a refresh token with an optional expiration time. create_email_token(data): Create an email verification token with a fixed expiration time. decode_token(token): Decode a JWT token and validate its scope. get_authenticated_user(token, db): Get the authenticated user based on the provided token. get_email_from_token(token): Get the email address from an email verification token.
Example Usage:: auth = Auth() hashed_password = auth.get_password_hash(“my_password”)

ALGORITHM = 'HS256'

SECRET_KEY = 'secret_key'

async allow_rout(request: Request, db: AsyncSession = Depends(get_db)) → User

Authenticate a user for accessing a private route.

This method is used as a dependency to authenticate a user when accessing a private route.

Parameters:

request (Request) – The request object, containing the user’s access token in cookies.
db (AsyncSession) – The asynchronous database session (Dependency).

Returns:

The authenticated user.

Return type:

User

Raises:

HTTPException 401 – Unauthorized if the user is not authenticated or the access token is missing.

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

Create an access token.

This function generates an access token with an optional expiration time.

Parameters:

data – dict: The data to include in the token payload.
expires_delta – int | None: The token’s expiration time (in minutes).

Returns:

The encoded access token.

Return type:

str

create_email_token(data: dict)

Create an email verification token.

This function generates an email verification token with an expiration time of 3 days.

Parameters:: data – dict: The data to include in the token payload.
Returns:: The encoded email verification token.
Return type:: str

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

Create a refresh token.

This function generates a refresh token with an optional expiration time.

Parameters:

data – dict: The data to include in the token payload.
expires_delta – float | None: The token’s expiration time (in seconds).

Returns:

The encoded refresh token.

Return type:

str

async decode_token(token: str)

Decode a JWT token.

This function decodes a JWT token and validates its scope.

Parameters:: token – str: The JWT token to decode.
Returns:: The payload of the decoded token.
Return type:: dict
Raises:: HTTPException – If the token is invalid.

async get_authenticated_user(token: str = Depends(OAuth2PasswordBearer), db: AsyncSession = Depends(get_db))

Get the authenticated user.

This function retrieves the currently authenticated user using the provided token.

Parameters:

token – str: The JWT token representing the user’s authentication.
db – AsyncSession: The database session.

Returns:

The authenticated user.

Return type:

User

Raises:

HTTPException – If the token is invalid or the user is not found.

async get_email_from_token(token: str)

Get the email address from an email verification token.

This function decodes an email verification token and retrieves the email address from it.

Parameters:: token – str: The JWT email verification token.
Returns:: The email address.
Return type:: str
Raises:: HTTPException – If the token is invalid.

get_password_hash(password: str)

Generate a hashed password from a plain text password.

Parameters:: password (str) – The plain text password.
Returns:: The hashed password.
Return type:: str

oauth2_scheme = <fastapi.security.oauth2.OAuth2PasswordBearer object>

pwd_context = <CryptContext>

property redis_cache

Get a Redis cache for storing user data.

This property returns a Redis cache for storing user data if it’s not already initialized.

Returns:: The Redis cache.
Return type:: aioredis.Redis

verify_password(plain_password, hashed_password)

Verify a plain password against a hashed password.

Parameters:

plain_password (str) – The plain text password.
hashed_password (str) – The hashed password to compare against.

Returns:

True if the plain password matches the hashed password, False otherwise.

Return type:

bool

Service Email

async src.services.email.reset_password_by_email(email: EmailStr, username: str, reset_token: str, host: str)

Send an email to reset a user’s password.

This function sends an email to the user with a link to reset their account password.

Parameters:

email – EmailStr: The recipient’s email address.
username – str: The username of the user receiving the email.
reset_token – str: The password reset token.
host – str: The hostname of your application for the reset link.

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

The send_email function sends an email to the user with a link to confirm their email address.

The function takes in three arguments:

email: the user’s email address, which is used as a unique identifier for them.
username: the username of the user who is registering. This will be displayed in their confirmation message so they know it was sent to them and not someone else.
host: this is where we are hosting our application, which will be used as part of our confirmation link.

Parameters:

email – EmailStr: Specify the email address of the recipient
username – str: Pass the username of the user to be sent in the email
host – str: Pass the hostname of your application to the template

Returns:

A coroutine object

Service Photos

src.services.photos.validate_crop_mode(crop_mode)

Validate Crop Mode

This function validates if a given crop mode is one of the allowed crop modes.

Parameters:: crop_mode (str) – The crop mode to be validated.
Returns:: True if the crop mode is valid.
Return type:: bool
Raises:: HTTPException 400 – If the crop mode is not in the list of allowed crop modes.

Example Usage:

                          validate_crop_mode('square')  # Returns True

                        

src.services.photos.validate_gravity_mode(gravity_mode)

Validate Gravity Mode

This function validates if a given gravity mode is one of the allowed gravity modes.

Parameters:: gravity_mode (str) – The gravity mode to be validated.
Returns:: True if the gravity mode is valid.
Return type:: bool
Raises:: HTTPException 400 – If the gravity mode is not in the list of allowed gravity modes.

Example Usage:

                          validate_gravity_mode('center')  # Returns True

                        

Service Roles

class src.services.roles.RoleChecker(allowed_roles: list[src.database.models.Role]): Bases: object

Schemas

class src.schemas.CommentList(*, limit: int = 10, offset: int = 0, photo_id: int)

Bases: BaseModel

Schema for listing comments.

limit: 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]] = {'limit': FieldInfo(annotation=int, required=False, default=10), 'offset': FieldInfo(annotation=int, required=False, default=0), 'photo_id': FieldInfo(annotation=int, 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.

offset: int

photo_id: int

class src.schemas.CommentRemoveSchema(*, id: int)

Bases: BaseModel

Schema for removing a comment.

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]] = {'id': FieldInfo(annotation=int, 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 src.schemas.CommentResponse(*, username: str, text: str, photo_id: int)

Bases: BaseModel

Schema for a comment response.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'photo_id': FieldInfo(annotation=int, required=True), 'text': FieldInfo(annotation=str, required=True), 'username': 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.

photo_id: int

text: str

username: str

class src.schemas.CommentSchema(*, text: str = 'some text', photo_id: int)

Bases: BaseModel

Schema for creating a comment.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'photo_id': FieldInfo(annotation=int, required=True), 'text': FieldInfo(annotation=str, required=False, default='some text')}

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

This replaces Model.__fields__ from Pydantic V1.

photo_id: int

text: str

class src.schemas.CommentUpdateSchema(*, id: int, text: str)

Bases: BaseModel

Schema for updating a comment.

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]] = {'id': FieldInfo(annotation=int, required=True), 'text': 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.

text: str

class src.schemas.MessageResponseSchema(*, message: str = 'This is a message')

Bases: BaseModel

Schema for a generic message response.

message: 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]] = {'message': FieldInfo(annotation=str, required=False, default='This is a message')}

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 src.schemas.PhotoRating(*, content: str)

Bases: BaseModel

Schema for adding a rating to a photo.

content: 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]] = {'content': 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 src.schemas.PhotosDb(*, id: int, url: str, description: str | None, user_id: int, created_at: datetime)

Bases: BaseModel

Schema for photo data retrieved from the database.

created_at: datetime

description: str | None

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), 'description': FieldInfo(annotation=Union[str, NoneType], required=True), 'id': FieldInfo(annotation=int, required=True), 'url': FieldInfo(annotation=str, required=True), 'user_id': FieldInfo(annotation=int, 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.

url: str

user_id: int

class src.schemas.RatingSchema(*, user_id: int, rating: int, photo_id: int)

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'photo_id': FieldInfo(annotation=int, required=True), 'rating': FieldInfo(annotation=int, required=True), 'user_id': FieldInfo(annotation=int, 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.

photo_id: int

rating: int

user_id: int

class src.schemas.RequestEmail(*, email: EmailStr)

Bases: BaseModel

Schema for requesting an email.

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 src.schemas.RequestRole(*, email: EmailStr, role: Role)

Bases: BaseModel

Schema for requesting a user role change.

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), 'role': FieldInfo(annotation=Role, 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.

role: Role

class src.schemas.TokenSchema(*, access_token: str, refresh_token: str, token_type: str = 'bearer')

Bases: BaseModel

Schema for a token response.

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=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.

refresh_token: str

token_type: str

class src.schemas.UserDb(*, id: int, username: str, email: str, avatar: str | None, role: Role, created_at: datetime, description: str | None)

Bases: BaseModel

Schema for user data retrieved from the database.

avatar: str | None

created_at: datetime

description: str | None

email: str

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]] = {'avatar': FieldInfo(annotation=Union[str, NoneType], required=True), 'created_at': FieldInfo(annotation=datetime, required=True), 'description': FieldInfo(annotation=Union[str, NoneType], required=True), 'email': FieldInfo(annotation=str, required=True), 'id': FieldInfo(annotation=int, required=True), 'role': FieldInfo(annotation=Role, required=True), 'username': 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.

role: Role

username: str

class src.schemas.UserProfileSchema(*, id: int, username: str, email: EmailStr, role: Role, avatar: str | None, photos_count: int | None, comments_count: int | None, is_active: bool | None, created_at: datetime)

Bases: BaseModel

Schema for user profile data.

avatar: str | None

comments_count: int | None

created_at: datetime

email: EmailStr

id: int

is_active: bool | 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]] = {'avatar': FieldInfo(annotation=Union[str, NoneType], required=True), 'comments_count': FieldInfo(annotation=Union[int, NoneType], required=True), 'created_at': FieldInfo(annotation=datetime, required=True), 'email': FieldInfo(annotation=EmailStr, required=True), 'id': FieldInfo(annotation=int, required=True), 'is_active': FieldInfo(annotation=Union[bool, NoneType], required=True), 'photos_count': FieldInfo(annotation=Union[int, NoneType], required=True), 'role': FieldInfo(annotation=Role, required=True), 'username': 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.

photos_count: int | None

role: Role

username: str

class src.schemas.UserResponseSchema(*, id: int, username: str, email: str, is_active: bool | None, created_at: datetime)

Bases: BaseModel

Schema for user response data.

created_at: datetime

email: str

id: int

is_active: bool | 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]] = {'created_at': FieldInfo(annotation=datetime, required=True), 'email': FieldInfo(annotation=str, required=True), 'id': FieldInfo(annotation=int, required=True), 'is_active': FieldInfo(annotation=Union[bool, NoneType], required=True), 'username': 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.

username: str

class src.schemas.UserSchema(*, username: str, email: EmailStr, password: str)

Bases: BaseModel

Schema for user registration data.

email: EmailStr

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'description': 'Schema for user registration data', 'example': {'email': 'example@example.com', 'password': 'qwer1234', 'username': 'sergiokapone'}, 'title': 'User Schema'}}: 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), 'password': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=6), MaxLen(max_length=30)]), 'username': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=5), MaxLen(max_length=25)])}

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

This replaces Model.__fields__ from Pydantic V1.

password: str

username: str

class src.schemas.UserUpdateSchema(*, username: str)

Bases: BaseModel

Schema for updating user data.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'username': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=5), MaxLen(max_length=25)])}

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

This replaces Model.__fields__ from Pydantic V1.

username: str

Welcome to SnapshotExchange’s documentation!

Database

REST API routes Auth

REST API routes Users

REST API routes Photos

REST API routes Comments

REST API routes Ratings

REST API routes Search

VIEWS routes Dashboard

VIEWS routes Photos

VIEWS routes Users

VIEWS routes Chat

Repository Users

Repository Photos

Repository Comments

Repository Ratings

Repository Search

Service Auth

Service Email

Service Photos

Service Roles

Schemas

Indices and tables