321 lines
9.5 KiB
Python
321 lines
9.5 KiB
Python
from random import randrange
|
||
|
||
from fastapi import APIRouter, Depends, status
|
||
from app.dependencies import get_current_user
|
||
from app.services import auth_service, user_service
|
||
from app.users.exceptions import (
|
||
PasswordAlreadyInUseException,
|
||
UserAlreadyExistsException,
|
||
UserNotFoundException,
|
||
)
|
||
from app.users.schemas import (
|
||
Responses,
|
||
SGetUsersFilter,
|
||
STokenLogin,
|
||
SUser,
|
||
SUserAvatars,
|
||
SUserChangeData,
|
||
SUserCode,
|
||
SUserFilter,
|
||
SUserLogin,
|
||
SUserPassword,
|
||
SUserRegister,
|
||
SUserResponse,
|
||
SUsers,
|
||
SUserSendConfirmationCode,
|
||
)
|
||
from app.utils.unit_of_work import UnitOfWork
|
||
|
||
router = APIRouter(prefix="/users", tags=["Пользователи"])
|
||
|
||
|
||
@router.get(
|
||
"",
|
||
status_code=status.HTTP_200_OK,
|
||
response_model=SUsers,
|
||
responses={
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR: {
|
||
"model": Responses.SBlackPhoenixException,
|
||
"description": "Внутренняя ошибка сервера"
|
||
},
|
||
},
|
||
)
|
||
async def get_all_users(filter_by=Depends(SGetUsersFilter), uow=Depends(UnitOfWork)):
|
||
users = await user_service.find_all(uow=uow, username=filter_by.username)
|
||
return users
|
||
|
||
|
||
@router.post(
|
||
"/check_existing_user",
|
||
status_code=status.HTTP_200_OK,
|
||
response_model=None,
|
||
responses={
|
||
status.HTTP_409_CONFLICT: {
|
||
"model": Responses.SUserAlreadyExistsException,
|
||
"description": "Пользователь с таким ником или почтой уже существует"
|
||
},
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR: {
|
||
"model": Responses.SBlackPhoenixException,
|
||
"description": "Внутренняя ошибка сервера"
|
||
},
|
||
},
|
||
)
|
||
async def check_existing_user(user_filter: SUserFilter, uow=Depends(UnitOfWork)):
|
||
try:
|
||
await user_service.find_user(uow=uow, **user_filter.model_dump(exclude_none=True))
|
||
raise UserAlreadyExistsException
|
||
except UserNotFoundException:
|
||
pass
|
||
|
||
|
||
@router.post(
|
||
"/check_existing_password",
|
||
status_code=status.HTTP_200_OK,
|
||
response_model=None,
|
||
responses={
|
||
status.HTTP_409_CONFLICT: {
|
||
"model": Responses.SPasswordAlreadyInUseException,
|
||
"description": "Пароль уже занят"
|
||
},
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR: {
|
||
"model": Responses.SBlackPhoenixException,
|
||
"description": "Внутренняя ошибка сервера"
|
||
},
|
||
},
|
||
)
|
||
async def check_existing_password(_: SUserPassword):
|
||
random_int = randrange(10)
|
||
if random_int > 5:
|
||
raise PasswordAlreadyInUseException
|
||
|
||
|
||
@router.post(
|
||
"/register",
|
||
status_code=status.HTTP_201_CREATED,
|
||
response_model=STokenLogin,
|
||
responses={
|
||
status.HTTP_409_CONFLICT: {
|
||
"model": Responses.SPasswordsMismatchException,
|
||
"description": "Пароли не совпадают"
|
||
},
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR: {
|
||
"model": Responses.SBlackPhoenixException,
|
||
"description": "Внутренняя ошибка сервера"
|
||
},
|
||
},
|
||
)
|
||
async def register_user(user_data: SUserRegister, uow=Depends(UnitOfWork)):
|
||
user = await user_service.add_user(uow=uow, user_data=user_data)
|
||
await user_service.send_confirmation_email(user=user, mail_type="registration")
|
||
|
||
access_token = auth_service.create_access_token({"sub": str(user.id)})
|
||
return {"authorization": f"Bearer {access_token}"}
|
||
|
||
|
||
@router.post(
|
||
"/resend_email_verification",
|
||
status_code=status.HTTP_200_OK,
|
||
response_model=None,
|
||
responses={
|
||
status.HTTP_401_UNAUTHORIZED: {
|
||
"model": Responses.STokenMissingException,
|
||
"description": "Токен отсутствует"
|
||
},
|
||
status.HTTP_403_FORBIDDEN: {
|
||
"model": Responses.SNotAuthenticated,
|
||
"description": "Not authenticated"
|
||
},
|
||
status.HTTP_404_NOT_FOUND: {
|
||
"model": Responses.SUserNotFoundException,
|
||
"description": "Юзер не найден"
|
||
},
|
||
status.HTTP_409_CONFLICT: {
|
||
"model": Responses.SIncorrectDataException,
|
||
"description": "Ты передал какую-то хуйню"
|
||
},
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR: {
|
||
"model": Responses.SBlackPhoenixException,
|
||
"description": "Внутренняя ошибка сервера"
|
||
},
|
||
},
|
||
)
|
||
async def resend_email_verification(user: SUser = Depends(get_current_user)):
|
||
await user_service.send_confirmation_email(user=user, mail_type="registration")
|
||
|
||
|
||
@router.post(
|
||
"/email_verification",
|
||
status_code=status.HTTP_200_OK,
|
||
response_model=None,
|
||
responses={
|
||
status.HTTP_401_UNAUTHORIZED: {
|
||
"model": Responses.STokenMissingException,
|
||
"description": "Токен отсутствует"
|
||
},
|
||
status.HTTP_403_FORBIDDEN: {
|
||
"model": Responses.SNotAuthenticated,
|
||
"description": "Not authenticated"
|
||
},
|
||
status.HTTP_404_NOT_FOUND: {
|
||
"model": Responses.SUserNotFoundException,
|
||
"description": "Юзер не найден"
|
||
},
|
||
status.HTTP_409_CONFLICT: {
|
||
"model": Responses.SWrongCodeException,
|
||
"description": "Введён не верный код подтверждения"
|
||
},
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR: {
|
||
"model": Responses.SBlackPhoenixException,
|
||
"description": "Внутренняя ошибка сервера"
|
||
},
|
||
},
|
||
)
|
||
async def email_verification(user_code: SUserCode, user: SUser = Depends(get_current_user), uow=Depends(UnitOfWork)):
|
||
confirmation_code = auth_service.get_confirmation_code(user_code.user_code)
|
||
await user_service.verificate_user(uow=uow, user=user, confirmation_code=confirmation_code)
|
||
|
||
|
||
@router.post(
|
||
"/login",
|
||
status_code=status.HTTP_200_OK,
|
||
response_model=STokenLogin,
|
||
responses={
|
||
status.HTTP_401_UNAUTHORIZED: {
|
||
"model": Responses.SIncorrectAuthDataException,
|
||
"description": "Введены не верные данные"
|
||
},
|
||
status.HTTP_409_CONFLICT: {
|
||
"model": Responses.SIncorrectDataException,
|
||
"description": "Ты передал какую-то хуйню"
|
||
},
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR: {
|
||
"model": Responses.SBlackPhoenixException,
|
||
"description": "Внутренняя ошибка сервера"
|
||
},
|
||
},
|
||
)
|
||
async def login_user(user_data: SUserLogin, uow=Depends(UnitOfWork)):
|
||
user = await auth_service.authenticate_user(
|
||
uow=uow,
|
||
email_or_username=user_data.email_or_username,
|
||
password=user_data.password,
|
||
)
|
||
access_token = auth_service.create_access_token({"sub": str(user.id)})
|
||
return {"authorization": f"Bearer {access_token}"}
|
||
|
||
|
||
@router.get(
|
||
"/me",
|
||
status_code=status.HTTP_200_OK,
|
||
response_model=SUserResponse,
|
||
responses={
|
||
status.HTTP_401_UNAUTHORIZED: {
|
||
"model": Responses.STokenMissingException,
|
||
"description": "Токен отсутствует"
|
||
},
|
||
status.HTTP_403_FORBIDDEN: {
|
||
"model": Responses.SNotAuthenticated,
|
||
"description": "Not authenticated"
|
||
},
|
||
status.HTTP_404_NOT_FOUND: {
|
||
"model": Responses.SUserNotFoundException,
|
||
"description": "Юзер не найден"
|
||
},
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR: {
|
||
"model": Responses.SBlackPhoenixException,
|
||
"description": "Внутренняя ошибка сервера"
|
||
},
|
||
},
|
||
)
|
||
async def get_user(user: SUser = Depends(get_current_user)):
|
||
return user
|
||
|
||
|
||
@router.get(
|
||
"/avatars",
|
||
status_code=status.HTTP_200_OK,
|
||
response_model=SUserAvatars,
|
||
responses={
|
||
status.HTTP_401_UNAUTHORIZED: {
|
||
"model": Responses.STokenMissingException,
|
||
"description": "Токен отсутствует"
|
||
},
|
||
status.HTTP_403_FORBIDDEN: {
|
||
"model": Responses.SNotAuthenticated,
|
||
"description": "Not authenticated"
|
||
},
|
||
status.HTTP_404_NOT_FOUND: {
|
||
"model": Responses.SUserNotFoundException,
|
||
"description": "Юзер не найден"
|
||
},
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR: {
|
||
"model": Responses.SBlackPhoenixException,
|
||
"description": "Внутренняя ошибка сервера"
|
||
},
|
||
},
|
||
)
|
||
async def get_user_avatars_history(user=Depends(get_current_user), uow=Depends(UnitOfWork)):
|
||
user_avatars = await user_service.get_user_avatars_history(uow=uow, user_id=user.id)
|
||
return user_avatars
|
||
|
||
|
||
@router.post(
|
||
"/send_confirmation_code",
|
||
status_code=status.HTTP_200_OK,
|
||
response_model=None,
|
||
responses={
|
||
status.HTTP_401_UNAUTHORIZED: {
|
||
"model": Responses.STokenMissingException,
|
||
"description": "Токен отсутствует"
|
||
},
|
||
status.HTTP_403_FORBIDDEN: {
|
||
"model": Responses.SNotAuthenticated,
|
||
"description": "Not authenticated"
|
||
},
|
||
status.HTTP_404_NOT_FOUND: {
|
||
"model": Responses.SUserNotFoundException,
|
||
"description": "Юзер не найден"
|
||
},
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR: {
|
||
"model": Responses.SBlackPhoenixException,
|
||
"description": "Внутренняя ошибка сервера"
|
||
},
|
||
},
|
||
)
|
||
async def send_confirmation_code(user_data: SUserSendConfirmationCode, user: SUser = Depends(get_current_user)):
|
||
await user_service.send_confirmation_email(user=user, mail_type="data_change_confirmation", email=user_data.email)
|
||
|
||
|
||
@router.post(
|
||
"/change_data",
|
||
status_code=status.HTTP_200_OK,
|
||
response_model=None,
|
||
responses={
|
||
status.HTTP_401_UNAUTHORIZED: {
|
||
"model": Responses.STokenMissingException,
|
||
"description": "Токен отсутствует"
|
||
},
|
||
status.HTTP_403_FORBIDDEN: {
|
||
"model": Responses.SNotAuthenticated,
|
||
"description": "Not authenticated"
|
||
},
|
||
status.HTTP_404_NOT_FOUND: {
|
||
"model": Responses.SUserNotFoundException,
|
||
"description": "Юзер не найден"
|
||
},
|
||
status.HTTP_409_CONFLICT: {
|
||
"model": Responses.SWrongCodeException,
|
||
"description": "Введён не верный код подтверждения"
|
||
},
|
||
status.HTTP_500_INTERNAL_SERVER_ERROR: {
|
||
"model": Responses.SBlackPhoenixException,
|
||
"description": "Внутренняя ошибка сервера"
|
||
},
|
||
},
|
||
)
|
||
async def change_user_data(
|
||
user_data: SUserChangeData,
|
||
user: SUser = Depends(get_current_user),
|
||
uow=Depends(UnitOfWork),
|
||
):
|
||
await user_service.change_data(uow=uow, user=user, user_data=user_data)
|