backend/src/service/user.py - master - nollieundergrob/PythonPackageStorage

backend/src/service/user.py
127 строк4 KB
nollieundergrob
SPLIT
13 мар 2025, 23:11
13 мар 2025, 23:11ad00f30
# import fake.user as data
# from routers.user import User
from schemas.user import User as Schema
from data.models.user import UserCRUD as data
from schemas.user import UserCreate, UserModify
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update, or_
from data.models.user import Users


class UserService:
    """READ"""
    @staticmethod
    async def get_all(session: AsyncSession) -> list[Schema]:
        users = await data.get_all(session=session)
        return [Schema.model_validate(user) for user in users]

    @staticmethod
    async def get_detail(user_id: int, session: AsyncSession) -> Schema:
        user = await data.get_detail(session=session, user_id=user_id)
        return Schema.model_validate(user) if user else None

    """CREATE"""

    @staticmethod
    async def create(session: AsyncSession, user_in: UserCreate) -> Users:

        existing_user = await session.execute(
            select(Users).where(Users.username == user_in.username)
        )
        if existing_user.scalar():
            raise ValueError("Username already exists")

        existing_email = await session.execute(
            select(Users).where(Users.email == user_in.email)
        )
        if existing_email.scalar():
            raise ValueError("Email already exists")

        user = Users(**user_in.model_dump())
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

    @staticmethod
    async def modify(session: AsyncSession, user_id: int, user_in: UserModify) -> Users:
        existing_user = await session.execute(select(Users).where(Users.id == user_id))
        if not existing_user.scalar():
            raise ValueError("Not Found")

        user_data = user_in.model_dump(exclude_unset=True)

        if "username" in user_data or "email" in user_data:
            conditions = []
            if "username" in user_data:
                conditions.append(Users.username == user_data["username"])
            if "email" in user_data:
                conditions.append(Users.email == user_data["email"])

            stmt = select(Users).where(or_(*conditions), Users.id != user_id)
            result = await session.execute(stmt)
            if result.scalar():
                raise ValueError("Username or email already exists")

        if user_data:
            await session.execute(
                update(Users).where(Users.id == user_id).values(**user_data)
            )
            await session.commit()

        updated_user = await session.execute(select(Users).where(Users.id == user_id))
        return updated_user.scalar_one()

    @staticmethod
    async def replace(
        session: AsyncSession, user_id: int, user_in: UserCreate
    ) -> Users:

        existing_user = await session.execute(select(Users).where(Users.id == user_id))
        if not existing_user.scalar():
            raise ValueError("Not Found")

        user_data = user_in.model_dump()

        stmt = select(Users).where(
            (Users.username == user_data["username"])
            | (Users.email == user_data["email"]),
            Users.id != user_id,
        )
        result = await session.execute(stmt)
        if result.scalar():
            raise ValueError("Username or email already exists")

        await session.execute(
            update(Users).where(Users.id == user_id).values(**user_data)
        )
        await session.commit()
        updated_user = await session.execute(select(Users).where(Users.id == user_id))
        return updated_user.scalar_one()

    @staticmethod
    async def delete(
        session: AsyncSession,
        user_id: int,
        current_user: Users
    ) -> Users:
        # Получаем пользователя для удаления
        result = await session.execute(
            select(Users).where(Users.id == user_id)
        )
        user_to_delete = result.scalar_one_or_none()

        if not user_to_delete:
            raise ValueError("User not found")

        if current_user.id != user_to_delete.id and not current_user.is_admin:
            raise PermissionError("Forbiden")

        # Удаляем пользователя
        await session.delete(user_to_delete)
        await session.commit()

        return user_to_delete


service = UserService()