51 lines
1.3 KiB
Python
51 lines
1.3 KiB
Python
"""User management service."""
|
|
|
|
from typing import Optional
|
|
from aiogram.types import User as TgUser
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from bot.db.operations import get_or_create_user, get_user_by_tg_id
|
|
from bot.db.models import User
|
|
from bot.logging_config import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class UserService:
|
|
"""Service for user-related operations."""
|
|
|
|
@staticmethod
|
|
async def ensure_user_exists(session: AsyncSession, tg_user: TgUser) -> User:
|
|
"""
|
|
Ensure user exists in database, create if not.
|
|
|
|
Args:
|
|
session: Database session
|
|
tg_user: Telegram user object
|
|
|
|
Returns:
|
|
User instance
|
|
"""
|
|
user = await get_or_create_user(
|
|
session=session,
|
|
tg_user_id=tg_user.id,
|
|
username=tg_user.username,
|
|
first_name=tg_user.first_name,
|
|
last_name=tg_user.last_name,
|
|
)
|
|
return user
|
|
|
|
@staticmethod
|
|
async def get_user(session: AsyncSession, tg_user_id: int) -> Optional[User]:
|
|
"""
|
|
Get user by Telegram ID.
|
|
|
|
Args:
|
|
session: Database session
|
|
tg_user_id: Telegram user ID
|
|
|
|
Returns:
|
|
User instance or None
|
|
"""
|
|
return await get_user_by_tg_id(session, tg_user_id)
|