Coverage for src/repository/users.py: 100%
42 statements
« prev ^ index » next coverage.py v7.5.4, created at 2024-07-01 22:29 +0200
1from fastapi import Depends, HTTPException, status
2from sqlalchemy import select
3from sqlalchemy.ext.asyncio import AsyncSession
5from libgravatar import Gravatar
7from src.database.db import get_db
8from src.entity.models import User
9from src.schemas.user import UserSchema
12async def get_user_by_email(email: str, db: AsyncSession = Depends(get_db)):
13 """
14 Retrieves a user from the database by their email address.
16 Args:
17 - email (str): The email address of the user to retrieve.
18 - db (AsyncSession, optional): The database session to use for the query. Defaults to Depends(get_db).
20 Returns:
21 - User | None: The user object if found, otherwise None.
23 Raises:
24 - HTTPException: If an error occurs while retrieving the user.
25 """
26 stmt = select(User).filter_by(email=email)
27 user = await db.execute(stmt)
28 user = user.scalar_one_or_none()
29 return user
32async def create_user(body: UserSchema, db: AsyncSession = Depends(get_db)):
33 """
34 Creates a new user in the database.
36 Args:
37 - body (UserSchema): The user schema containing the user's data.
38 - db (AsyncSession, optional): The database session to use for the query. Defaults to Depends(get_db).
40 Returns:
41 - User: The newly created user object.
43 Raises:
44 - HTTPException: If an error occurs while creating the user.
46 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.
47 """
48 avatar = None
49 try:
50 avatar = Gravatar(body.email).get_image()
51 except Exception as error:
52 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error)
54 new_user = User(**body.model_dump(), avatar=avatar)
55 db.add(new_user)
56 await db.commit()
57 await db.refresh(new_user)
58 return new_user
61async def update_token(user: User, refresh_token: str | None, db: AsyncSession):
62 """
63 Updates the refresh token for the given user in the database.
65 Args:
66 - user (User): The user object to update the refresh token for.
67 - refresh_token (str | None): The new refresh token to set for the user, or None to remove the refresh token.
68 - db (AsyncSession): The database session to use for the update.
70 Returns:
71 - None: This function does not return a value.
73 Raises:
74 - None: This function does not raise any exceptions.
76 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.
77 """
78 user.refresh_token = refresh_token
79 await db.commit()
82async def confirmed_email(email: str, db: AsyncSession):
83 """
84 Confirms the email for the given user in the database.
86 Args:
87 - email (str): The email address of the user to confirm.
88 - db (AsyncSession): The database session to use for the update.
90 Returns:
91 - None: This function does not return a value.
93 Raises:
94 - None: This function does not raise any exceptions.
96 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.
97 """
98 user = await get_user_by_email(email, db)
99 user.confirmed = True
100 await db.commit()
101 await db.refresh(user)
104async def update_avatar(email, url: str, db: AsyncSession) -> User:
105 """
106 Updates the avatar URL for the given user in the database.
108 Args:
109 - email (str): The email address of the user to update the avatar for.
110 - url (str): The new avatar URL to set for the user.
111 - db (AsyncSession): The database session to use for the update.
113 Returns:
114 - User: The updated user object with the new avatar URL.
116 Raises:
117 - None: This function does not raise any exceptions.
119 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.
120 """
121 user = await get_user_by_email(email, db)
122 user.avatar = url
123 db.commit()
124 return user
127async def update_password(user: User, new_password: str, db: AsyncSession):
128 """
129 Updates the password for the given user in the database.
131 Args:
132 - user (User): The user object to update the password for.
133 - new_password (str): The new password to set for the user.
134 - db (AsyncSession): The database session to use for the update.
136 Returns:
137 - User: The updated user object with the new password.
139 Raises:
140 - None: This function does not raise any exceptions.
142 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.
143 """
144 async with db as session:
145 user.password = new_password
146 await session.commit()
147 await session.refresh(user)
148 return user