127 lines
5.0 KiB
Python
127 lines
5.0 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import joinedload
|
|
from sqlalchemy import and_
|
|
from models import Assessment, ClassGroup, Exercise
|
|
from .base_repository import BaseRepository
|
|
|
|
|
|
class AssessmentRepository(BaseRepository[Assessment]):
|
|
"""Repository pour les évaluations."""
|
|
|
|
def __init__(self):
|
|
super().__init__(Assessment)
|
|
|
|
def find_by_filters(
|
|
self,
|
|
trimester: Optional[int] = None,
|
|
class_id: Optional[int] = None,
|
|
correction_status: Optional[str] = None,
|
|
sort_by: str = 'date_desc'
|
|
) -> List[Assessment]:
|
|
"""Trouve les évaluations selon les filtres avec eager loading des classes."""
|
|
query = Assessment.query.options(
|
|
joinedload(Assessment.class_group)
|
|
)
|
|
|
|
# Application des filtres
|
|
filters = []
|
|
if trimester:
|
|
filters.append(Assessment.trimester == trimester)
|
|
if class_id:
|
|
filters.append(Assessment.class_group_id == class_id)
|
|
|
|
if filters:
|
|
query = query.filter(and_(*filters))
|
|
|
|
# Application du tri
|
|
query = self._apply_sorting(query, sort_by)
|
|
|
|
# Récupérer les résultats
|
|
assessments = query.all()
|
|
|
|
# Filtrer par statut de correction si nécessaire
|
|
if correction_status:
|
|
assessments = self._filter_by_correction_status(assessments, correction_status)
|
|
|
|
return assessments
|
|
|
|
def find_with_full_details(self, id: int) -> Optional[Assessment]:
|
|
"""Trouve une évaluation avec tous ses détails."""
|
|
return Assessment.query.options(
|
|
joinedload(Assessment.class_group),
|
|
joinedload(Assessment.exercises).joinedload(Exercise.grading_elements)
|
|
).filter_by(id=id).first()
|
|
|
|
def get_or_404(self, id: int) -> Assessment:
|
|
"""Récupère une évaluation ou lève une erreur 404."""
|
|
return Assessment.query.get_or_404(id)
|
|
|
|
def get_with_class_or_404(self, id: int) -> Assessment:
|
|
"""Récupère une évaluation avec sa classe ou lève une erreur 404."""
|
|
from flask import abort
|
|
assessment = Assessment.query.options(
|
|
joinedload(Assessment.class_group)
|
|
).filter_by(id=id).first()
|
|
if not assessment:
|
|
abort(404)
|
|
return assessment
|
|
|
|
def get_with_full_details_or_404(self, id: int) -> Assessment:
|
|
"""Récupère une évaluation avec tous ses détails ou lève une erreur 404."""
|
|
from flask import abort
|
|
assessment = self.find_with_full_details(id)
|
|
if not assessment:
|
|
abort(404)
|
|
return assessment
|
|
|
|
def find_recent(self, limit: int = 5) -> List[Assessment]:
|
|
"""Trouve les évaluations récentes."""
|
|
return Assessment.query.order_by(
|
|
Assessment.date.desc()
|
|
).limit(limit).all()
|
|
|
|
def find_by_class_group(self, class_group_id: int) -> List[Assessment]:
|
|
"""Trouve toutes les évaluations d'une classe."""
|
|
return Assessment.query.filter_by(
|
|
class_group_id=class_group_id
|
|
).order_by(Assessment.date.desc()).all()
|
|
|
|
def find_by_trimester(self, trimester: int) -> List[Assessment]:
|
|
"""Trouve toutes les évaluations d'un trimestre."""
|
|
return Assessment.query.filter_by(
|
|
trimester=trimester
|
|
).order_by(Assessment.date.desc()).all()
|
|
|
|
def count_by_class_group(self, class_group_id: int) -> int:
|
|
"""Compte les évaluations d'une classe."""
|
|
return Assessment.query.filter_by(class_group_id=class_group_id).count()
|
|
|
|
def _apply_sorting(self, query, sort_by: str):
|
|
"""Applique le tri à la requête."""
|
|
if sort_by == 'date_desc':
|
|
return query.order_by(Assessment.date.desc())
|
|
elif sort_by == 'date_asc':
|
|
return query.order_by(Assessment.date.asc())
|
|
elif sort_by == 'title':
|
|
return query.order_by(Assessment.title.asc())
|
|
elif sort_by == 'class':
|
|
return query.join(ClassGroup).order_by(ClassGroup.name.asc())
|
|
return query
|
|
|
|
def _filter_by_correction_status(self, assessments: List[Assessment], status: str) -> List[Assessment]:
|
|
"""Filtre les évaluations par statut de correction."""
|
|
filtered_assessments = []
|
|
|
|
for assessment in assessments:
|
|
progress = assessment.grading_progress
|
|
progress_status = progress.get('status', 'not_started')
|
|
|
|
# Mapper les statuts de progression aux filtres
|
|
if status == 'complete' and progress_status == 'completed':
|
|
filtered_assessments.append(assessment)
|
|
elif status == 'incomplete' and progress_status in ['in_progress', 'not_started']:
|
|
filtered_assessments.append(assessment)
|
|
elif status == 'not_started' and progress_status == 'not_started':
|
|
filtered_assessments.append(assessment)
|
|
|
|
return filtered_assessments |