334 lines
12 KiB
Python
334 lines
12 KiB
Python
"""
|
||
Репозитории для работы с моделями базы данных системы тестирования
|
||
"""
|
||
|
||
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
|
||
|
||
|
||
class TestRepository:
|
||
"""Репозиторий для работы с тестами"""
|
||
|
||
def __init__(self, session: Session):
|
||
self.session = session
|
||
|
||
def create_test(
|
||
self,
|
||
title: str,
|
||
description: str = None,
|
||
category_id: int = None,
|
||
is_active: bool = True,
|
||
time_limit_minutes: int = None
|
||
) -> 'Test':
|
||
"""Создание нового теста"""
|
||
from database.models import Test
|
||
|
||
test = Test(
|
||
title=title,
|
||
description=description,
|
||
category_id=category_id,
|
||
is_active=is_active,
|
||
time_limit_minutes=time_limit_minutes
|
||
)
|
||
self.session.add(test)
|
||
self.session.commit()
|
||
self.session.refresh(test)
|
||
return test
|
||
|
||
def get_test_by_id(self, test_id: int) -> Optional['Test']:
|
||
"""Получение теста по ID"""
|
||
from database.models import Test
|
||
return self.session.query(Test).filter(Test.id == test_id).first()
|
||
|
||
def get_tests(
|
||
self,
|
||
is_active: bool = None,
|
||
category_id: int = None,
|
||
limit: int = None
|
||
) -> List['Test']:
|
||
"""Получение списка тестов с фильтрацией"""
|
||
from database.models import Test
|
||
|
||
query = self.session.query(Test)
|
||
|
||
if is_active is not None:
|
||
query = query.filter(Test.is_active == is_active)
|
||
|
||
if category_id is not None:
|
||
query = query.filter(Test.category_id == category_id)
|
||
|
||
if limit:
|
||
query = query.limit(limit)
|
||
|
||
return query.all()
|
||
|
||
def update_test(self, test_id: int, **kwargs) -> Optional['Test']:
|
||
"""Обновление теста"""
|
||
from database.models import Test
|
||
|
||
test = self.session.query(Test).filter(Test.id == test_id).first()
|
||
if test:
|
||
for key, value in kwargs.items():
|
||
if hasattr(test, key):
|
||
setattr(test, key, value)
|
||
self.session.commit()
|
||
self.session.refresh(test)
|
||
return test
|
||
|
||
def delete_test(self, test_id: int) -> bool:
|
||
"""Удаление теста"""
|
||
from database.models import Test
|
||
|
||
test = self.session.query(Test).filter(Test.id == test_id).first()
|
||
if test:
|
||
self.session.delete(test)
|
||
self.session.commit()
|
||
return True
|
||
return False
|
||
|
||
|
||
class QuestionRepository:
|
||
"""Репозиторий для работы с вопросами"""
|
||
|
||
def __init__(self, session: Session):
|
||
self.session = session
|
||
|
||
def create_question(
|
||
self,
|
||
test_id: int,
|
||
question_text: str,
|
||
question_type: str = "single_choice",
|
||
order_number: int = 1,
|
||
points: int = 1,
|
||
image_id: int = None
|
||
) -> 'Question':
|
||
"""Создание нового вопроса"""
|
||
from database.models import Question
|
||
|
||
question = Question(
|
||
test_id=test_id,
|
||
question_text=question_text,
|
||
question_type=question_type,
|
||
order_number=order_number,
|
||
points=points,
|
||
image_id=image_id
|
||
)
|
||
self.session.add(question)
|
||
self.session.commit()
|
||
self.session.refresh(question)
|
||
return question
|
||
|
||
def get_question_by_id(self, question_id: int) -> Optional['Question']:
|
||
"""Получение вопроса по ID"""
|
||
from database.models import Question
|
||
return self.session.query(Question).filter(Question.id == question_id).first()
|
||
|
||
def get_questions_by_test_id(self, test_id: int) -> List['Question']:
|
||
"""Получение всех вопросов для теста"""
|
||
from database.models import Question
|
||
return self.session.query(Question).filter(
|
||
Question.test_id == test_id
|
||
).order_by(Question.order_number).all()
|
||
|
||
def update_question(self, question_id: int, **kwargs) -> Optional['Question']:
|
||
"""Обновление вопроса"""
|
||
from database.models import Question
|
||
|
||
question = self.session.query(Question).filter(Question.id == question_id).first()
|
||
if question:
|
||
for key, value in kwargs.items():
|
||
if hasattr(question, key):
|
||
setattr(question, key, value)
|
||
self.session.commit()
|
||
self.session.refresh(question)
|
||
return question
|
||
|
||
def delete_question(self, question_id: int) -> bool:
|
||
"""Удаление вопроса"""
|
||
from database.models import Question
|
||
|
||
question = self.session.query(Question).filter(Question.id == question_id).first()
|
||
if question:
|
||
self.session.delete(question)
|
||
self.session.commit()
|
||
return True
|
||
return False |