60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
|
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()
|