refactor: extract duplicated patterns into shared helpers
Backend: create api/helpers.py with eligible_enrollment_filter, count_eligible_students, get_active_enrollment, ensure_unique_name, upsert_app_configs, and build_heatmap. Add full_name properties to Student model. Apply across all route files (-481/+184 lines). Frontend: create stores/helpers.js with withLoading composable, apply to assessments and classes Pinia stores. 96/96 tests pass. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
196
backend/api/helpers.py
Normal file
196
backend/api/helpers.py
Normal file
@@ -0,0 +1,196 @@
|
||||
"""
|
||||
Fonctions utilitaires partagées entre les routes API.
|
||||
Élimine la duplication de code pour les patterns courants.
|
||||
"""
|
||||
|
||||
from datetime import date
|
||||
from typing import Optional, Dict, List, Tuple, Any, Callable
|
||||
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy import select, func
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from infrastructure.database.models import (
|
||||
AppConfig,
|
||||
StudentEnrollment,
|
||||
Student,
|
||||
)
|
||||
from schemas.assessment import HeatmapCell, HeatmapData
|
||||
|
||||
|
||||
def eligible_enrollment_filter(class_group_id: int, at_date: date):
|
||||
"""
|
||||
Retourne les clauses WHERE pour filtrer les inscriptions actives à une date donnée.
|
||||
|
||||
Usage:
|
||||
query = select(StudentEnrollment).where(
|
||||
*eligible_enrollment_filter(class_id, assessment.date)
|
||||
)
|
||||
"""
|
||||
return (
|
||||
StudentEnrollment.class_group_id == class_group_id,
|
||||
StudentEnrollment.enrollment_date <= at_date,
|
||||
(
|
||||
StudentEnrollment.departure_date.is_(None)
|
||||
| (StudentEnrollment.departure_date >= at_date)
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
async def count_eligible_students(
|
||||
session: AsyncSession, class_group_id: int, at_date: date
|
||||
) -> int:
|
||||
"""Compte les élèves inscrits dans une classe à une date donnée."""
|
||||
query = select(func.count(StudentEnrollment.id)).where(
|
||||
*eligible_enrollment_filter(class_group_id, at_date)
|
||||
)
|
||||
result = await session.execute(query)
|
||||
return result.scalar() or 0
|
||||
|
||||
|
||||
def eligible_student_ids_subquery(class_group_id: int, at_date: date):
|
||||
"""Retourne une sous-requête des student_id éligibles à une date donnée."""
|
||||
return select(StudentEnrollment.student_id).where(
|
||||
*eligible_enrollment_filter(class_group_id, at_date)
|
||||
)
|
||||
|
||||
|
||||
async def get_eligible_enrollments(
|
||||
session: AsyncSession, class_group_id: int, at_date: date
|
||||
):
|
||||
"""Récupère les inscriptions éligibles avec les étudiants chargés."""
|
||||
query = (
|
||||
select(StudentEnrollment)
|
||||
.options(selectinload(StudentEnrollment.student))
|
||||
.where(*eligible_enrollment_filter(class_group_id, at_date))
|
||||
)
|
||||
result = await session.execute(query)
|
||||
return result.scalars().all()
|
||||
|
||||
|
||||
def get_active_enrollment(
|
||||
student: Student,
|
||||
) -> Optional[StudentEnrollment]:
|
||||
"""Retourne l'inscription active (sans date de départ) d'un élève."""
|
||||
for enrollment in student.enrollments:
|
||||
if enrollment.departure_date is None:
|
||||
return enrollment
|
||||
return None
|
||||
|
||||
|
||||
async def ensure_unique_name(
|
||||
session: AsyncSession,
|
||||
model_class,
|
||||
name: str,
|
||||
*,
|
||||
field_name: str = "name",
|
||||
exclude_id: Optional[int] = None,
|
||||
entity_label: str = "enregistrement",
|
||||
):
|
||||
"""
|
||||
Vérifie qu'un nom est unique pour un modèle donné.
|
||||
Lève HTTPException 400 si un doublon est trouvé.
|
||||
"""
|
||||
field = getattr(model_class, field_name)
|
||||
query = select(model_class).where(field == name)
|
||||
|
||||
if exclude_id is not None:
|
||||
query = query.where(model_class.id != exclude_id)
|
||||
|
||||
result = await session.execute(query)
|
||||
if result.scalar_one_or_none():
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"Un(e) {entity_label} avec le nom '{name}' existe déjà",
|
||||
)
|
||||
|
||||
|
||||
async def upsert_app_configs(
|
||||
session: AsyncSession, updates: Dict[str, str]
|
||||
) -> None:
|
||||
"""
|
||||
Met à jour ou crée des entrées AppConfig en lot.
|
||||
Ne fait PAS de commit — l'appelant doit appeler session.commit().
|
||||
"""
|
||||
for key, value in updates.items():
|
||||
query = select(AppConfig).where(AppConfig.key == key)
|
||||
result = await session.execute(query)
|
||||
config = result.scalar_one_or_none()
|
||||
|
||||
if config:
|
||||
config.value = value
|
||||
else:
|
||||
session.add(AppConfig(key=key, value=value))
|
||||
|
||||
|
||||
def build_heatmap(
|
||||
enrollments,
|
||||
assessment,
|
||||
items: dict,
|
||||
item_extractor: Callable,
|
||||
grading_calc,
|
||||
sorted_student_names: List[str],
|
||||
*,
|
||||
color_map: Optional[Dict[str, str]] = None,
|
||||
) -> Optional[HeatmapData]:
|
||||
"""
|
||||
Construit une HeatmapData pour des compétences ou domaines.
|
||||
|
||||
Args:
|
||||
enrollments: liste d'inscriptions avec .student chargé
|
||||
assessment: l'évaluation avec .exercises chargés
|
||||
items: set ou dict des items à évaluer (noms)
|
||||
item_extractor: fonction (element) -> nom de l'item ou None
|
||||
grading_calc: instance de GradingCalculator
|
||||
sorted_student_names: noms triés pour l'axe Y
|
||||
color_map: optionnel, dict nom -> couleur pour les cellules
|
||||
"""
|
||||
if not items:
|
||||
return None
|
||||
|
||||
item_names = set(items) if isinstance(items, dict) else items
|
||||
cells = []
|
||||
|
||||
for enrollment in enrollments:
|
||||
student = enrollment.student
|
||||
student_name = student.full_name_reversed
|
||||
|
||||
item_scores = {name: {"score": 0.0, "max": 0.0} for name in item_names}
|
||||
|
||||
for exercise in assessment.exercises:
|
||||
for element in exercise.grading_elements:
|
||||
item_name = item_extractor(element)
|
||||
if item_name and item_name in item_scores:
|
||||
for g in element.grades:
|
||||
if g.student_id == student.id and g.value:
|
||||
calc_score = grading_calc.calculate_score(
|
||||
g.value.strip(),
|
||||
element.grading_type,
|
||||
element.max_points,
|
||||
)
|
||||
if calc_score is not None:
|
||||
item_scores[item_name]["score"] += calc_score
|
||||
item_scores[item_name]["max"] += element.max_points
|
||||
break
|
||||
|
||||
for name, data in item_scores.items():
|
||||
if data["max"] > 0:
|
||||
pct = round((data["score"] / data["max"]) * 100, 1)
|
||||
cell_kwargs = dict(
|
||||
student_id=student.id,
|
||||
student_name=student_name,
|
||||
item_name=name,
|
||||
score=round(data["score"], 2),
|
||||
max_points=data["max"],
|
||||
percentage=pct,
|
||||
)
|
||||
if color_map and name in color_map:
|
||||
cell_kwargs["color"] = color_map[name]
|
||||
cells.append(HeatmapCell(**cell_kwargs))
|
||||
|
||||
return HeatmapData(
|
||||
items=sorted(list(item_names)),
|
||||
students=sorted_student_names,
|
||||
cells=cells,
|
||||
)
|
||||
Reference in New Issue
Block a user