feat: add temporal student gestion
This commit is contained in:
		
							
								
								
									
										166
									
								
								repositories/temporal_student_repository.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										166
									
								
								repositories/temporal_student_repository.py
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,166 @@ | ||||
| from typing import List, Optional, Tuple | ||||
| from datetime import date | ||||
| from sqlalchemy import and_, or_ | ||||
| from models import db, Student, StudentEnrollment, Assessment | ||||
| from repositories.base_repository import BaseRepository | ||||
|  | ||||
|  | ||||
| class TemporalStudentRepository(BaseRepository[Student]): | ||||
|     """Repository pour gérer les étudiants avec logique temporelle.""" | ||||
|      | ||||
|     def __init__(self): | ||||
|         super().__init__(Student) | ||||
|      | ||||
|     def find_enrolled_in_class_at_date(self, class_group_id: int, check_date: date) -> List[Student]: | ||||
|         """Trouve les étudiants inscrits dans une classe à une date donnée.""" | ||||
|         from sqlalchemy import func | ||||
|         return db.session.query(Student)\ | ||||
|             .join(StudentEnrollment)\ | ||||
|             .filter( | ||||
|                 StudentEnrollment.class_group_id == class_group_id, | ||||
|                 StudentEnrollment.enrollment_date <= check_date, | ||||
|                 or_( | ||||
|                     StudentEnrollment.departure_date.is_(None), | ||||
|                     StudentEnrollment.departure_date >= check_date | ||||
|                 ) | ||||
|             )\ | ||||
|             .order_by(func.lower(Student.last_name), func.lower(Student.first_name))\ | ||||
|             .all() | ||||
|      | ||||
|     def find_eligible_for_assessment(self, assessment: Assessment) -> List[Student]: | ||||
|         """Trouve les étudiants éligibles pour une évaluation donnée.""" | ||||
|         if not assessment.date: | ||||
|             return [] | ||||
|          | ||||
|         return self.find_enrolled_in_class_at_date(assessment.class_group_id, assessment.date) | ||||
|      | ||||
|     def find_current_students_in_class(self, class_group_id: int) -> List[Student]: | ||||
|         """Trouve les étudiants actuellement inscrits dans une classe.""" | ||||
|         from sqlalchemy import func | ||||
|         return db.session.query(Student)\ | ||||
|             .join(StudentEnrollment)\ | ||||
|             .filter( | ||||
|                 StudentEnrollment.class_group_id == class_group_id, | ||||
|                 StudentEnrollment.departure_date.is_(None) | ||||
|             )\ | ||||
|             .order_by(func.lower(Student.last_name), func.lower(Student.first_name))\ | ||||
|             .all() | ||||
|      | ||||
|     def get_enrollment_history(self, student_id: int) -> List[StudentEnrollment]: | ||||
|         """Récupère l'historique complet des inscriptions d'un élève.""" | ||||
|         return StudentEnrollment.query\ | ||||
|             .filter_by(student_id=student_id)\ | ||||
|             .order_by(StudentEnrollment.enrollment_date.desc())\ | ||||
|             .all() | ||||
|      | ||||
|     def find_students_with_movements_in_period(self, start_date: date, end_date: date) -> List[Tuple[Student, List[StudentEnrollment]]]: | ||||
|         """Trouve les étudiants qui ont eu des mouvements (arrivée/départ) dans une période avec leurs inscriptions.""" | ||||
|         # Récupérer les étudiants qui ont eu des mouvements dans la période | ||||
|         students_with_movements = db.session.query(Student)\ | ||||
|             .join(StudentEnrollment)\ | ||||
|             .filter( | ||||
|                 or_( | ||||
|                     # Arrivées dans la période | ||||
|                     and_( | ||||
|                         StudentEnrollment.enrollment_date >= start_date, | ||||
|                         StudentEnrollment.enrollment_date <= end_date | ||||
|                     ), | ||||
|                     # Départs dans la période | ||||
|                     and_( | ||||
|                         StudentEnrollment.departure_date >= start_date, | ||||
|                         StudentEnrollment.departure_date <= end_date | ||||
|                     ) | ||||
|                 ) | ||||
|             )\ | ||||
|             .distinct()\ | ||||
|             .order_by(Student.last_name, Student.first_name)\ | ||||
|             .all() | ||||
|          | ||||
|         # Pour chaque étudiant, récupérer ses mouvements dans la période | ||||
|         result = [] | ||||
|         for student in students_with_movements: | ||||
|             movements = StudentEnrollment.query\ | ||||
|                 .filter_by(student_id=student.id)\ | ||||
|                 .filter( | ||||
|                     or_( | ||||
|                         # Arrivées dans la période | ||||
|                         and_( | ||||
|                             StudentEnrollment.enrollment_date >= start_date, | ||||
|                             StudentEnrollment.enrollment_date <= end_date | ||||
|                         ), | ||||
|                         # Départs dans la période | ||||
|                         and_( | ||||
|                             StudentEnrollment.departure_date >= start_date, | ||||
|                             StudentEnrollment.departure_date <= end_date | ||||
|                         ) | ||||
|                     ) | ||||
|                 )\ | ||||
|                 .order_by(StudentEnrollment.enrollment_date.desc())\ | ||||
|                 .all() | ||||
|              | ||||
|             if movements: | ||||
|                 result.append((student, movements)) | ||||
|          | ||||
|         return result | ||||
|      | ||||
|     def create_enrollment(self, student_id: int, class_group_id: int,  | ||||
|                          enrollment_date: date, enrollment_reason: str = None) -> StudentEnrollment: | ||||
|         """Crée une nouvelle inscription pour un élève.""" | ||||
|         # Vérifier s'il y a déjà une inscription active | ||||
|         active_enrollment = StudentEnrollment.query.filter_by( | ||||
|             student_id=student_id, | ||||
|             departure_date=None | ||||
|         ).first() | ||||
|          | ||||
|         if active_enrollment: | ||||
|             raise ValueError("L'élève a déjà une inscription active") | ||||
|          | ||||
|         # Créer la nouvelle inscription | ||||
|         enrollment = StudentEnrollment( | ||||
|             student_id=student_id, | ||||
|             class_group_id=class_group_id, | ||||
|             enrollment_date=enrollment_date, | ||||
|             enrollment_reason=enrollment_reason | ||||
|         ) | ||||
|          | ||||
|         db.session.add(enrollment) | ||||
|         return enrollment | ||||
|      | ||||
|     def end_enrollment(self, student_id: int, departure_date: date,  | ||||
|                       departure_reason: str = None) -> Optional[StudentEnrollment]: | ||||
|         """Termine l'inscription active d'un élève.""" | ||||
|         active_enrollment = StudentEnrollment.query.filter_by( | ||||
|             student_id=student_id, | ||||
|             departure_date=None | ||||
|         ).first() | ||||
|          | ||||
|         if not active_enrollment: | ||||
|             return None | ||||
|          | ||||
|         active_enrollment.departure_date = departure_date | ||||
|         active_enrollment.departure_reason = departure_reason | ||||
|          | ||||
|         return active_enrollment | ||||
|      | ||||
|     def transfer_student(self, student_id: int, new_class_group_id: int,  | ||||
|                         transfer_date: date, transfer_reason: str = None) -> tuple[StudentEnrollment, StudentEnrollment]: | ||||
|         """Transfère un élève d'une classe à une autre.""" | ||||
|         # Terminer l'inscription actuelle | ||||
|         old_enrollment = self.end_enrollment( | ||||
|             student_id,  | ||||
|             transfer_date,  | ||||
|             f"Transfert: {transfer_reason}" if transfer_reason else "Transfert" | ||||
|         ) | ||||
|          | ||||
|         if not old_enrollment: | ||||
|             raise ValueError("Aucune inscription active trouvée pour cet élève") | ||||
|          | ||||
|         # Créer la nouvelle inscription | ||||
|         new_enrollment = self.create_enrollment( | ||||
|             student_id, | ||||
|             new_class_group_id, | ||||
|             transfer_date, | ||||
|             f"Transfert: {transfer_reason}" if transfer_reason else "Transfert" | ||||
|         ) | ||||
|          | ||||
|         return old_enrollment, new_enrollment | ||||
		Reference in New Issue
	
	Block a user