94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
from app.config import settings
|
|
from app.services.redis_service import get_redis_session, RedisService
|
|
from app.tasks.tasks import generate_confirmation_code, send_confirmation_email, send_data_change_email
|
|
from app.utils.unit_of_work import UnitOfWork
|
|
from app.users.exceptions import PasswordsMismatchException, WrongCodeException
|
|
from app.users.schemas import SUser, SUsers, SUserRegister, SConfirmationData, SUserAvatars, SUserChangeData
|
|
from app.services.auth_service import AuthService
|
|
|
|
|
|
class UserService:
|
|
@staticmethod
|
|
async def find_user(uow: UnitOfWork, **find_by) -> SUser:
|
|
async with uow:
|
|
user = await uow.user.find_one(**find_by)
|
|
return user
|
|
|
|
@staticmethod
|
|
async def find_all(uow: UnitOfWork, username: str) -> SUsers:
|
|
async with uow:
|
|
users = await uow.user.find_all(username=username)
|
|
return users
|
|
|
|
@staticmethod
|
|
async def add_user(uow: UnitOfWork, user_data: SUserRegister) -> SUser:
|
|
if user_data.password != user_data.password2:
|
|
raise PasswordsMismatchException
|
|
|
|
hashed_password = AuthService.get_password_hash(user_data.password)
|
|
async with uow:
|
|
user = await uow.user.add(
|
|
email=user_data.email,
|
|
hashed_password=hashed_password,
|
|
username=user_data.username,
|
|
date_of_birth=user_data.date_of_birth,
|
|
)
|
|
await uow.user.add_user_avatar(user_id=user.id, avatar=str(user.avatar_image))
|
|
await uow.commit()
|
|
return user
|
|
|
|
@staticmethod
|
|
async def send_confirmation_email(user: SUser, mail_type: str, email: str | None = None) -> None:
|
|
redis_session = get_redis_session()
|
|
user_code = generate_confirmation_code()
|
|
await RedisService.delete_verification_code(redis=redis_session, user_id=user.id)
|
|
await RedisService.set_verification_code(redis=redis_session, user_id=user.id, verification_code=user_code)
|
|
|
|
user_mail_data = SConfirmationData.model_validate(
|
|
{
|
|
"user_id": user.id,
|
|
"username": user.username,
|
|
"email_to": email or user.email,
|
|
"confirmation_code": user_code,
|
|
"type": mail_type,
|
|
}
|
|
)
|
|
send_confirmation_email.delay(user_mail_data.model_dump())
|
|
|
|
@staticmethod
|
|
async def verificate_user(uow: UnitOfWork, user: SUser, confirmation_code: str) -> None:
|
|
redis_session = get_redis_session()
|
|
verification_code = await RedisService.get_verification_code(redis=redis_session, user_id=user.id)
|
|
|
|
if verification_code != confirmation_code:
|
|
raise WrongCodeException
|
|
async with uow:
|
|
await uow.user.change_data(user_id=user.id, role=settings.VERIFICATED_USER)
|
|
await uow.commit()
|
|
|
|
@staticmethod
|
|
async def get_user_avatars_history(uow: UnitOfWork, user_id: int) -> SUserAvatars:
|
|
async with uow:
|
|
user_avatars = await uow.user.get_user_avatars(user_id=user_id)
|
|
return user_avatars
|
|
|
|
@staticmethod
|
|
async def change_data(uow: UnitOfWork, user: SUser, user_data: SUserChangeData) -> None:
|
|
redis_session = get_redis_session()
|
|
verification_code = await RedisService.get_verification_code(redis=redis_session, user_id=user.id)
|
|
if verification_code != user_data.verification_code:
|
|
raise WrongCodeException
|
|
hashed_password = AuthService.get_password_hash(user_data.new_password) if user_data.new_password else user.hashed_password
|
|
async with uow:
|
|
await uow.user.change_data(
|
|
user_id=user.id,
|
|
email=user_data.email,
|
|
username=user_data.username,
|
|
avatar_image=str(user_data.avatar_url or user.avatar_image),
|
|
hashed_password=hashed_password
|
|
)
|
|
(await uow.user.add_user_avatar(user_id=user.id, avatar=str(user_data.avatar_url))) if user_data.avatar_url else None
|
|
await uow.commit()
|
|
|
|
send_data_change_email.delay(user_data.username, user_data.email)
|
|
await RedisService.delete_verification_code(redis=redis_session, user_id=user.id)
|