backend/src/api/services/user.py - master - FlysAt/DataFoundry

backend/src/api/services/user.py
69 строк3 KB
AlexanderShmygol
fixed fullname param
16 май 2026, 17:24
16 май 2026, 17:24aeff8a6
from fastapi import HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from src.api.repositories.user import UserRepository
from src.api.services.auth import AuthService
from src.core.security import hash_password
from src.models.user import User
from src.schemas.user import UserInfoUpdateSchema


class UserService:
    def __init__(self, db: AsyncSession) -> None:
        self.users = UserRepository(db)

    async def get_user(self, user_id: int) -> User:
        return await self._get_existing_user(user_id)

    async def update_user(self, user_id: int, payload: UserInfoUpdateSchema) -> User:
        user = await self._get_existing_user(user_id)
        self._validate_roles(payload.roles)
        updated_fields = getattr(payload, "model_fields_set", None)
        if updated_fields is None:
            updated_fields = payload.__fields_set__

        if "email" in updated_fields and payload.email is None:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Email cannot be null",
            )
        if "email" in updated_fields and payload.email != user.email:
            existing_user = await self.users.get_by_email(payload.email)
            if existing_user and existing_user.id != user.id:
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail="User already exists",
            )
            user.email = payload.email

        if "password" in updated_fields and payload.password is not None:
            user.hashed_password = hash_password(payload.password)
        if "full_name" in updated_fields:
            user.full_name = payload.full_name
        if "rating" in updated_fields and payload.rating is not None:
            user.rating = payload.rating
        if "roles" in updated_fields and payload.roles is not None:
            user.roles = payload.roles or ["user"]

        return await self.users.update(user)

    async def delete_user(self, user_id: int) -> None:
        user = await self._get_existing_user(user_id)
        await self.users.delete(user)

    async def _get_existing_user(self, user_id: int) -> User:
        user = await self.users.get_by_id(user_id)
        if not user:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
        return user

    @staticmethod
    def _validate_roles(roles: list[str] | None) -> None:
        if roles is None:
            return
        invalid_roles = set(roles) - AuthService.ALLOWED_ROLES
        if invalid_roles:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"Invalid roles: {', '.join(invalid_roles)}",
            )