tg_bot_language/services/user_service.py

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from database.models import User, LanguageLevel
from typing import Optional


class UserService:
    """Сервис для работы с пользователями"""

    @staticmethod
    async def get_or_create_user(session: AsyncSession, telegram_id: int, username: Optional[str] = None) -> User:
        """
        Получить пользователя или создать нового

        Args:
            session: Сессия базы данных
            telegram_id: Telegram ID пользователя
            username: Username пользователя

        Returns:
            Объект пользователя
        """
        # Попытка найти существующего пользователя
        result = await session.execute(
            select(User).where(User.telegram_id == telegram_id)
        )
        user = result.scalar_one_or_none()

        if user:
            return user

        # Создание нового пользователя
        new_user = User(
            telegram_id=telegram_id,
            username=username,
            level=LanguageLevel.A1
        )
        session.add(new_user)
        await session.commit()
        await session.refresh(new_user)

        return new_user

    @staticmethod
    async def get_user_by_telegram_id(session: AsyncSession, telegram_id: int) -> Optional[User]:
        """
        Получить пользователя по Telegram ID

        Args:
            session: Сессия базы данных
            telegram_id: Telegram ID пользователя

        Returns:
            Объект пользователя или None
        """
        result = await session.execute(
            select(User).where(User.telegram_id == telegram_id)
        )
        return result.scalar_one_or_none()

    @staticmethod
    async def update_user_level(session: AsyncSession, user_id: int, level: LanguageLevel):
        """
        Обновить уровень английского пользователя

        Args:
            session: Сессия базы данных
            user_id: ID пользователя
            level: Новый уровень
        """
        result = await session.execute(
            select(User).where(User.id == user_id)
        )
        user = result.scalar_one_or_none()

        if user:
            user.level = level
            await session.commit()

    @staticmethod
    async def update_user_language(session: AsyncSession, user_id: int, language: str):
        """
        Обновить язык интерфейса пользователя

        Args:
            session: Сессия базы данных
            user_id: ID пользователя
            language: Новый язык (ru/en)
        """
        result = await session.execute(
            select(User).where(User.id == user_id)
        )
        user = result.scalar_one_or_none()

        if user:
            user.language_interface = language
            await session.commit()

    @staticmethod
    async def update_user_learning_language(session: AsyncSession, user_id: int, language: str):
        """
        Обновить язык изучения пользователя

        Args:
            session: Сессия базы данных
            user_id: ID пользователя
            language: Новый язык изучения (ISO2)
        """
        result = await session.execute(
            select(User).where(User.id == user_id)
        )
        user = result.scalar_one_or_none()

        if user:
            user.learning_language = language
            await session.commit()