bd - api
This commit is contained in:
185
database/repositories.py
Normal file
185
database/repositories.py
Normal file
@@ -0,0 +1,185 @@
|
||||
"""
|
||||
Репозитории для работы с моделями базы данных системы тестирования
|
||||
"""
|
||||
|
||||
from typing import Optional, List
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import and_, or_
|
||||
from database.models import (
|
||||
User, Message, BotSettings, Test, TestCategory, Question,
|
||||
AnswerOption, TestResult, UserAnswer, Image
|
||||
)
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class UserRepository:
|
||||
"""Репозиторий для работы с пользователями"""
|
||||
|
||||
@staticmethod
|
||||
def create_user(session: Session, telegram_id: int, username: str = None,
|
||||
first_name: str = None, last_name: str = None) -> User:
|
||||
"""Создание нового пользователя"""
|
||||
user = User(
|
||||
telegram_id=telegram_id,
|
||||
username=username,
|
||||
first_name=first_name,
|
||||
last_name=last_name
|
||||
)
|
||||
session.add(user)
|
||||
session.flush() # Получаем ID без коммита
|
||||
return user
|
||||
|
||||
@staticmethod
|
||||
def get_user_by_telegram_id(session: Session, telegram_id: int) -> Optional[User]:
|
||||
"""Получение пользователя по Telegram ID"""
|
||||
return session.query(User).filter(User.telegram_id == telegram_id).first()
|
||||
|
||||
@staticmethod
|
||||
def get_user_by_id(session: Session, user_id: int) -> Optional[User]:
|
||||
"""Получение пользователя по ID"""
|
||||
return session.query(User).filter(User.id == user_id).first()
|
||||
|
||||
@staticmethod
|
||||
def get_all_users(session: Session, active_only: bool = True) -> List[User]:
|
||||
"""Получение всех пользователей"""
|
||||
query = session.query(User)
|
||||
if active_only:
|
||||
query = query.filter(User.is_active == True)
|
||||
return query.all()
|
||||
|
||||
@staticmethod
|
||||
def update_user(session: Session, telegram_id: int, **kwargs) -> Optional[User]:
|
||||
"""Обновление данных пользователя"""
|
||||
user = UserRepository.get_user_by_telegram_id(session, telegram_id)
|
||||
if user:
|
||||
for key, value in kwargs.items():
|
||||
if hasattr(user, key):
|
||||
setattr(user, key, value)
|
||||
session.flush()
|
||||
return user
|
||||
|
||||
@staticmethod
|
||||
def get_or_create_user(session: Session, telegram_id: int, **kwargs) -> tuple[User, bool]:
|
||||
"""Получение или создание пользователя. Возвращает (user, created)"""
|
||||
user = UserRepository.get_user_by_telegram_id(session, telegram_id)
|
||||
if user:
|
||||
return user, False
|
||||
else:
|
||||
user = UserRepository.create_user(session, telegram_id, **kwargs)
|
||||
return user, True
|
||||
|
||||
class MessageRepository:
|
||||
"""Репозиторий для работы с сообщениями"""
|
||||
|
||||
@staticmethod
|
||||
def create_message(session: Session, telegram_message_id: int, user_id: int,
|
||||
text: str = None, message_type: str = 'text') -> Message:
|
||||
"""Создание нового сообщения"""
|
||||
message = Message(
|
||||
telegram_message_id=telegram_message_id,
|
||||
user_id=user_id,
|
||||
text=text,
|
||||
message_type=message_type
|
||||
)
|
||||
session.add(message)
|
||||
session.flush()
|
||||
return message
|
||||
|
||||
@staticmethod
|
||||
def get_messages_by_user(session: Session, user_id: int, limit: int = 100) -> List[Message]:
|
||||
"""Получение сообщений пользователя"""
|
||||
return (session.query(Message)
|
||||
.filter(Message.user_id == user_id)
|
||||
.order_by(Message.created_at.desc())
|
||||
.limit(limit)
|
||||
.all())
|
||||
|
||||
@staticmethod
|
||||
def get_recent_messages(session: Session, limit: int = 100) -> List[Message]:
|
||||
"""Получение последних сообщений"""
|
||||
return (session.query(Message)
|
||||
.order_by(Message.created_at.desc())
|
||||
.limit(limit)
|
||||
.all())
|
||||
|
||||
class SettingsRepository:
|
||||
"""Репозиторий для работы с настройками бота"""
|
||||
|
||||
@staticmethod
|
||||
def get_setting(session: Session, key: str) -> Optional[str]:
|
||||
"""Получение значения настройки"""
|
||||
setting = session.query(BotSettings).filter(BotSettings.key == key).first()
|
||||
return setting.value if setting else None
|
||||
|
||||
@staticmethod
|
||||
def set_setting(session: Session, key: str, value: str, description: str = None) -> BotSettings:
|
||||
"""Установка значения настройки"""
|
||||
setting = session.query(BotSettings).filter(BotSettings.key == key).first()
|
||||
if setting:
|
||||
setting.value = value
|
||||
if description:
|
||||
setting.description = description
|
||||
else:
|
||||
setting = BotSettings(key=key, value=value, description=description)
|
||||
session.add(setting)
|
||||
session.flush()
|
||||
return setting
|
||||
|
||||
@staticmethod
|
||||
def get_all_settings(session: Session) -> List[BotSettings]:
|
||||
"""Получение всех настроек"""
|
||||
return session.query(BotSettings).all()
|
||||
|
||||
@staticmethod
|
||||
def delete_setting(session: Session, key: str) -> bool:
|
||||
"""Удаление настройки"""
|
||||
setting = session.query(BotSettings).filter(BotSettings.key == key).first()
|
||||
if setting:
|
||||
session.delete(setting)
|
||||
session.flush()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class ImageRepository:
|
||||
"""Репозиторий для работы с изображениями"""
|
||||
|
||||
@staticmethod
|
||||
def create_image(session: Session, filename: str, content_type: str,
|
||||
image_data: bytes, alt_text: str = None) -> Image:
|
||||
"""Создание записи изображения"""
|
||||
image = Image(
|
||||
filename=filename,
|
||||
content_type=content_type,
|
||||
file_size=len(image_data),
|
||||
image_data=image_data,
|
||||
alt_text=alt_text
|
||||
)
|
||||
session.add(image)
|
||||
session.flush()
|
||||
return image
|
||||
|
||||
@staticmethod
|
||||
def get_image_by_id(session: Session, image_id: int) -> Optional[Image]:
|
||||
"""Получение изображения по ID"""
|
||||
return session.query(Image).filter(Image.id == image_id).first()
|
||||
|
||||
@staticmethod
|
||||
def get_image_by_filename(session: Session, filename: str) -> Optional[Image]:
|
||||
"""Получение изображения по имени файла"""
|
||||
return session.query(Image).filter(Image.filename == filename).first()
|
||||
|
||||
@staticmethod
|
||||
def get_all_images(session: Session) -> List[Image]:
|
||||
"""Получение всех изображений"""
|
||||
return session.query(Image).all()
|
||||
|
||||
@staticmethod
|
||||
def delete_image(session: Session, image_id: int) -> bool:
|
||||
"""Удаление изображения"""
|
||||
image = session.query(Image).filter(Image.id == image_id).first()
|
||||
if image:
|
||||
session.delete(image)
|
||||
return True
|
||||
return False
|
||||
Reference in New Issue
Block a user