130 lines
5.4 KiB
Python
130 lines
5.4 KiB
Python
from typing import List, Optional, Dict, Any
|
|
from sqlalchemy.orm import joinedload
|
|
from models import Grade, GradingElement, Exercise, Assessment, Student
|
|
from .base_repository import BaseRepository
|
|
|
|
|
|
class GradeRepository(BaseRepository[Grade]):
|
|
"""Repository pour les notes."""
|
|
|
|
def __init__(self):
|
|
super().__init__(Grade)
|
|
|
|
def find_by_student_and_element(self, student_id: int, grading_element_id: int) -> Optional[Grade]:
|
|
"""Trouve une note par étudiant et élément de notation."""
|
|
return Grade.query.filter_by(
|
|
student_id=student_id,
|
|
grading_element_id=grading_element_id
|
|
).first()
|
|
|
|
def find_or_create_by_student_and_element(self, student_id: int, grading_element_id: int) -> Grade:
|
|
"""Trouve ou crée une note par étudiant et élément de notation."""
|
|
grade = self.find_by_student_and_element(student_id, grading_element_id)
|
|
if not grade:
|
|
grade = Grade(
|
|
student_id=student_id,
|
|
grading_element_id=grading_element_id
|
|
)
|
|
self.save(grade)
|
|
self.flush() # Pour obtenir l'ID
|
|
return grade
|
|
|
|
def find_by_assessment(self, assessment_id: int) -> List[Grade]:
|
|
"""Trouve toutes les notes d'une évaluation."""
|
|
return Grade.query.join(
|
|
GradingElement
|
|
).join(
|
|
Exercise
|
|
).filter_by(
|
|
assessment_id=assessment_id
|
|
).all()
|
|
|
|
def find_by_student(self, student_id: int) -> List[Grade]:
|
|
"""Trouve toutes les notes d'un étudiant."""
|
|
return Grade.query.filter_by(
|
|
student_id=student_id
|
|
).all()
|
|
|
|
def delete_by_student(self, student_id: int) -> int:
|
|
"""Supprime toutes les notes d'un étudiant. Retourne le nombre supprimé."""
|
|
count = Grade.query.filter_by(student_id=student_id).count()
|
|
Grade.query.filter_by(student_id=student_id).delete()
|
|
return count
|
|
|
|
def find_existing_grades_for_assessment(self, assessment_id: int) -> Dict[str, Grade]:
|
|
"""
|
|
Trouve les notes existantes d'une évaluation indexées par clé.
|
|
Clé format: "{student_id}_{grading_element_id}"
|
|
"""
|
|
existing_grades = {}
|
|
for grade in self.find_by_assessment(assessment_id):
|
|
key = f"{grade.student_id}_{grade.grading_element_id}"
|
|
existing_grades[key] = grade
|
|
return existing_grades
|
|
|
|
def bulk_update_or_create_grades(self, grade_data: List[Dict[str, Any]]) -> int:
|
|
"""
|
|
Met à jour ou crée plusieurs notes en lot.
|
|
|
|
Args:
|
|
grade_data: Liste de dictionnaires avec student_id, grading_element_id, value, comment
|
|
|
|
Returns:
|
|
Nombre de notes traitées
|
|
"""
|
|
count = 0
|
|
for data in grade_data:
|
|
grade = self.find_by_student_and_element(
|
|
data['student_id'],
|
|
data['grading_element_id']
|
|
)
|
|
|
|
if data.get('value') or data.get('comment'):
|
|
if not grade:
|
|
grade = Grade(
|
|
student_id=data['student_id'],
|
|
grading_element_id=data['grading_element_id'],
|
|
value=data.get('value'),
|
|
comment=data.get('comment')
|
|
)
|
|
self.save(grade)
|
|
else:
|
|
grade.value = data.get('value')
|
|
grade.comment = data.get('comment')
|
|
count += 1
|
|
elif grade:
|
|
# Supprimer si valeur et commentaire vides
|
|
self.delete(grade)
|
|
count += 1
|
|
|
|
return count
|
|
|
|
def find_by_student_trimester_with_elements(self, student_id: int, trimester: int) -> List[Grade]:
|
|
"""Trouve toutes les notes d'un élève pour un trimestre avec les éléments de notation."""
|
|
return Grade.query.join(GradingElement).join(Exercise).join(Assessment).filter(
|
|
Grade.student_id == student_id,
|
|
Assessment.trimester == trimester
|
|
).options(
|
|
joinedload(Grade.grading_element).joinedload(GradingElement.exercise).joinedload(Exercise.assessment),
|
|
joinedload(Grade.grading_element).joinedload(GradingElement.domain)
|
|
).all()
|
|
|
|
def find_by_class_trimester(self, class_group_id: int, trimester: int) -> List[Grade]:
|
|
"""Trouve toutes les notes d'une classe pour un trimestre."""
|
|
return Grade.query.join(Student).join(GradingElement).join(Exercise).join(Assessment).filter(
|
|
Student.class_group_id == class_group_id,
|
|
Assessment.trimester == trimester
|
|
).options(
|
|
joinedload(Grade.student),
|
|
joinedload(Grade.grading_element).joinedload(GradingElement.exercise).joinedload(Exercise.assessment),
|
|
joinedload(Grade.grading_element).joinedload(GradingElement.domain)
|
|
).all()
|
|
|
|
def get_student_grades_by_assessment(self, student_id: int, assessment_id: int) -> List[Grade]:
|
|
"""Récupère toutes les notes d'un élève pour une évaluation spécifique."""
|
|
return Grade.query.join(GradingElement).join(Exercise).filter(
|
|
Grade.student_id == student_id,
|
|
Exercise.assessment_id == assessment_id
|
|
).options(
|
|
joinedload(Grade.grading_element).joinedload(GradingElement.exercise)
|
|
).all() |