Migration v1 (Flask) -> v2 (FastAPI + Vue.js) complétée

 Changements majeurs:
- Suppression complète du code Flask legacy
- Migration backend FastAPI vers racine /backend
- Migration frontend Vue.js vers racine /frontend
- Suppression de notytex-v2/ (code monté à la racine)

 Validations:
- Backend démarre correctement (port 8000)
- API /api/v2/health répond healthy
- 99/99 tests unitaires passent
- Frontend configuré avec proxy Vite

📝 Documentation:
- README.md réécrit pour v2
- Instructions de démarrage mises à jour
- .gitignore adapté pour backend/frontend/

🎯 Architecture finale:
notytex/
├── backend/     # FastAPI + SQLAlchemy + Pydantic
├── frontend/    # Vue 3 + Vite + TailwindCSS
├── docs/        # Documentation
└── school_management.db  # Base de données (inchangée)

Jalon 6 complété: Application v2 prête pour utilisation!
This commit is contained in:
2025-11-25 21:09:47 +01:00
parent 60c60c1605
commit 2b08eb534a
4125 changed files with 303 additions and 453271 deletions

0
backend/api/__init__.py Normal file
View File

View File

@@ -0,0 +1,14 @@
"""
Dépendances FastAPI pour l'injection de dépendances.
"""
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from infrastructure.database.session import get_async_session
# Type alias pour la session async
AsyncSessionDep = Annotated[AsyncSession, Depends(get_async_session)]

111
backend/api/main.py Normal file
View File

@@ -0,0 +1,111 @@
"""
Application FastAPI principale pour Notytex v2.
"""
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import text, inspect
from core.config import settings
from infrastructure.database.session import async_engine, sync_engine
from infrastructure.database.models import (
ClassGroup,
Student,
Assessment,
Exercise,
GradingElement,
Grade,
)
# Import des routes
from api.routes import classes, students, assessments, config, council
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Gestion du cycle de vie de l'application.
"""
# Startup
print(f"Starting {settings.app_name} v{settings.app_version}")
yield
# Shutdown
await async_engine.dispose()
print("Application shutdown complete")
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
description="API REST pour le système de gestion scolaire Notytex",
docs_url=f"{settings.api_v2_prefix}/docs",
redoc_url=f"{settings.api_v2_prefix}/redoc",
openapi_url=f"{settings.api_v2_prefix}/openapi.json",
lifespan=lifespan,
)
# Configuration CORS
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get(f"{settings.api_v2_prefix}/health")
async def health_check():
"""
Endpoint de santé pour vérifier que l'API est opérationnelle
et connectée à la base de données.
Returns:
Dict avec statut, connexion DB et comptage des entités
"""
try:
# Utiliser une connexion sync pour l'inspection des tables
# car inspect() n'est pas async
with sync_engine.connect() as conn:
inspector = inspect(sync_engine)
tables = inspector.get_table_names()
# Compter les entités principales
classes_count = conn.execute(text("SELECT COUNT(*) FROM class_group")).scalar()
students_count = conn.execute(text("SELECT COUNT(*) FROM student")).scalar()
assessments_count = conn.execute(text("SELECT COUNT(*) FROM assessment")).scalar()
return {
"status": "healthy",
"database": "connected",
"tables": len(tables),
"classes": classes_count,
"students": students_count,
"assessments": assessments_count,
}
except Exception as e:
return {
"status": "unhealthy",
"database": "disconnected",
"error": str(e),
}
@app.get("/")
async def root():
"""Redirection vers la documentation API."""
return {
"message": f"Bienvenue sur {settings.app_name}",
"version": settings.app_version,
"docs": f"{settings.api_v2_prefix}/docs",
}
# Enregistrement des routes
app.include_router(classes.router, prefix=settings.api_v2_prefix)
app.include_router(students.router, prefix=settings.api_v2_prefix)
app.include_router(assessments.router, prefix=settings.api_v2_prefix)
app.include_router(config.router, prefix=settings.api_v2_prefix)
app.include_router(council.router, prefix=settings.api_v2_prefix)

View File

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,839 @@
"""
Routes API pour les classes (ClassGroup).
"""
import csv
from io import StringIO
from typing import Optional, List, Tuple
from datetime import date
from fastapi import APIRouter, HTTPException, Query, UploadFile, File, Form
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
from api.dependencies import AsyncSessionDep
from infrastructure.database.models import (
ClassGroup,
Student,
StudentEnrollment,
Assessment,
Exercise,
GradingElement,
Grade,
Domain,
Competence,
)
from schemas.class_group import (
ClassGroupRead,
ClassGroupDetail,
ClassGroupList,
ClassGroupCreate,
ClassGroupUpdate,
ClassGroupResponse,
ClassDashboardStats,
StudentAverage,
HistogramBin,
DomainStats,
CompetenceStats,
)
from domain.services.grading_calculator import GradingCalculator
from schemas.student import StudentWithClass, StudentList
from schemas.csv_import import (
CSVImportResponse,
ImportedStudentInfo,
SkippedStudentInfo,
ImportErrorInfo,
)
router = APIRouter(prefix="/classes", tags=["Classes"])
@router.get("", response_model=ClassGroupList)
async def get_classes(
session: AsyncSessionDep,
year: Optional[str] = Query(None, description="Filtrer par année scolaire"),
):
"""
Récupère la liste de toutes les classes.
"""
query = select(ClassGroup)
if year:
query = query.where(ClassGroup.year == year)
query = query.order_by(ClassGroup.year.desc(), ClassGroup.name)
result = await session.execute(query)
classes = result.scalars().all()
# Calculer le nombre d'étudiants et d'évaluations pour chaque classe
classes_with_count = []
for cls in classes:
# Compter les étudiants actuellement inscrits
students_count_query = select(func.count(StudentEnrollment.id)).where(
StudentEnrollment.class_group_id == cls.id,
StudentEnrollment.departure_date.is_(None)
)
students_count_result = await session.execute(students_count_query)
students_count = students_count_result.scalar() or 0
# Compter les évaluations de cette classe
assessments_count_query = select(func.count(Assessment.id)).where(
Assessment.class_group_id == cls.id
)
assessments_count_result = await session.execute(assessments_count_query)
assessments_count = assessments_count_result.scalar() or 0
classes_with_count.append(
ClassGroupRead(
id=cls.id,
name=cls.name,
description=cls.description,
year=cls.year,
students_count=students_count,
assessments_count=assessments_count
)
)
return ClassGroupList(
classes=classes_with_count,
total=len(classes_with_count)
)
@router.get("/{class_id}", response_model=ClassGroupDetail)
async def get_class(
class_id: int,
session: AsyncSessionDep,
):
"""
Récupère les détails d'une classe spécifique.
"""
query = select(ClassGroup).where(ClassGroup.id == class_id)
result = await session.execute(query)
cls = result.scalar_one_or_none()
if not cls:
raise HTTPException(status_code=404, detail="Classe non trouvée")
# Compter les étudiants
count_query = select(func.count(StudentEnrollment.id)).where(
StudentEnrollment.class_group_id == cls.id,
StudentEnrollment.departure_date.is_(None)
)
count_result = await session.execute(count_query)
students_count = count_result.scalar() or 0
# Compter les évaluations
assessments_query = select(func.count(Assessment.id)).where(
Assessment.class_group_id == cls.id
)
assessments_result = await session.execute(assessments_query)
assessments_count = assessments_result.scalar() or 0
return ClassGroupDetail(
id=cls.id,
name=cls.name,
description=cls.description,
year=cls.year,
students_count=students_count,
assessments_count=assessments_count
)
@router.get("/{class_id}/students", response_model=StudentList)
async def get_class_students(
class_id: int,
session: AsyncSessionDep,
include_departed: bool = Query(False, description="Inclure les élèves partis"),
at_date: Optional[str] = Query(None, description="Filtrer les élèves inscrits à cette date (YYYY-MM-DD)"),
):
"""
Récupère la liste des étudiants d'une classe.
Si at_date est fourni, retourne uniquement les élèves qui étaient inscrits à cette date.
"""
# Vérifier que la classe existe
class_query = select(ClassGroup).where(ClassGroup.id == class_id)
class_result = await session.execute(class_query)
cls = class_result.scalar_one_or_none()
if not cls:
raise HTTPException(status_code=404, detail="Classe non trouvée")
# Récupérer les inscriptions
query = (
select(StudentEnrollment)
.options(selectinload(StudentEnrollment.student))
.where(StudentEnrollment.class_group_id == class_id)
)
# Filtrer par date si spécifié
if at_date:
from datetime import date as date_type
try:
filter_date = date_type.fromisoformat(at_date)
except ValueError:
raise HTTPException(status_code=400, detail="Format de date invalide (YYYY-MM-DD)")
query = query.where(
StudentEnrollment.enrollment_date <= filter_date,
(StudentEnrollment.departure_date.is_(None) |
(StudentEnrollment.departure_date >= filter_date))
)
elif not include_departed:
query = query.where(StudentEnrollment.departure_date.is_(None))
result = await session.execute(query)
enrollments = result.scalars().all()
students = []
for enrollment in enrollments:
student = enrollment.student
students.append(
StudentWithClass(
id=student.id,
last_name=student.last_name,
first_name=student.first_name,
email=student.email,
full_name=f"{student.first_name} {student.last_name}",
current_class_id=class_id if enrollment.departure_date is None else None,
current_class_name=cls.name if enrollment.departure_date is None else None
)
)
# Trier par nom de famille puis prénom
students.sort(key=lambda s: (s.last_name.lower(), s.first_name.lower()))
return StudentList(
students=students,
total=len(students)
)
@router.get("/{class_id}/stats", response_model=ClassDashboardStats)
async def get_class_stats(
class_id: int,
trimester: int,
session: AsyncSessionDep,
):
"""
Récupère les statistiques complètes d'une classe pour un trimestre.
Inclut:
- Moyennes par élève
- Statistiques globales (moyenne, médiane, écart-type)
- Histogramme des moyennes
- Analyse par domaines et compétences
"""
# Vérifier que la classe existe
class_query = select(ClassGroup).where(ClassGroup.id == class_id)
class_result = await session.execute(class_query)
cls = class_result.scalar_one_or_none()
if not cls:
raise HTTPException(status_code=404, detail="Classe non trouvée")
# Récupérer les élèves actuellement inscrits
students_query = (
select(Student)
.join(StudentEnrollment, Student.id == StudentEnrollment.student_id)
.where(
StudentEnrollment.class_group_id == class_id,
StudentEnrollment.departure_date.is_(None)
)
.order_by(Student.last_name, Student.first_name)
)
students_result = await session.execute(students_query)
students = students_result.scalars().all()
# Récupérer les évaluations du trimestre
assessments_query = select(Assessment).where(
Assessment.class_group_id == class_id,
Assessment.trimester == trimester
)
assessments_result = await session.execute(assessments_query)
assessments = assessments_result.scalars().all()
# Calculer les moyennes de chaque élève
calculator = GradingCalculator()
student_averages = []
all_averages = []
for student in students:
weighted_sum = 0.0
total_coefficient = 0.0
assessment_count = 0
for assessment in assessments:
# Récupérer les notes de l'élève pour cette évaluation
grades_query = (
select(Grade, GradingElement)
.join(GradingElement, Grade.grading_element_id == GradingElement.id)
.join(Exercise, GradingElement.exercise_id == Exercise.id)
.where(
Exercise.assessment_id == assessment.id,
Grade.student_id == student.id
)
)
grades_result = await session.execute(grades_query)
grades_data = grades_result.all()
if grades_data:
total_score = 0.0
total_max_points = 0.0
for grade, element in grades_data:
if grade.value:
score = calculator.calculate_score(
grade.value, element.grading_type, element.max_points
)
if score is not None and calculator.is_counted_in_total(grade.value):
total_score += score
total_max_points += element.max_points
if total_max_points > 0:
# Ramener sur 20
score_on_20 = total_score / total_max_points * 20
weighted_sum += score_on_20 * assessment.coefficient
total_coefficient += assessment.coefficient
assessment_count += 1
# Calculer la moyenne pondérée
average = None
if total_coefficient > 0:
average = round(weighted_sum / total_coefficient, 2)
all_averages.append(average)
student_averages.append(StudentAverage(
student_id=student.id,
first_name=student.first_name,
last_name=student.last_name,
full_name=f"{student.first_name} {student.last_name}",
average=average,
assessment_count=assessment_count
))
# Calculer les statistiques globales
mean = median = std_dev = min_score = max_score = None
if all_averages:
mean = round(sum(all_averages) / len(all_averages), 2)
sorted_averages = sorted(all_averages)
n = len(sorted_averages)
if n % 2 == 0:
median = (sorted_averages[n // 2 - 1] + sorted_averages[n // 2]) / 2
else:
median = sorted_averages[n // 2]
median = round(median, 2)
min_score = min(all_averages)
max_score = max(all_averages)
# Écart-type
variance = sum((x - mean) ** 2 for x in all_averages) / len(all_averages)
std_dev = round(variance ** 0.5, 2)
# Créer l'histogramme (bins de 2 points)
histogram = []
for i in range(0, 20, 2):
count = sum(1 for avg in all_averages if i <= avg < i + 2)
histogram.append(HistogramBin(
range_start=float(i),
range_end=float(i + 2),
label=f"{i}-{i + 2}",
count=count
))
# Ajouter le dernier bin pour 20
count_20 = sum(1 for avg in all_averages if avg == 20)
if count_20 > 0:
histogram[-1].count += count_20
# Compter les évaluations par statut
assessments_completed = 0
assessments_in_progress = 0
for assessment in assessments:
# Compter les notes pour cette évaluation
grades_count_query = (
select(func.count(Grade.id))
.join(GradingElement, Grade.grading_element_id == GradingElement.id)
.join(Exercise, GradingElement.exercise_id == Exercise.id)
.where(Exercise.assessment_id == assessment.id)
)
grades_count_result = await session.execute(grades_count_query)
grades_count = grades_count_result.scalar() or 0
# Compter le total attendu
elements_count_query = (
select(func.count(GradingElement.id))
.join(Exercise, GradingElement.exercise_id == Exercise.id)
.where(Exercise.assessment_id == assessment.id)
)
elements_count_result = await session.execute(elements_count_query)
elements_count = elements_count_result.scalar() or 0
expected_grades = elements_count * len(students)
if expected_grades > 0:
if grades_count >= expected_grades:
assessments_completed += 1
elif grades_count > 0:
assessments_in_progress += 1
# Statistiques par domaine et compétence (simplifié)
domains_stats = []
competences_stats = []
# Récupérer les domaines
domains_query = select(Domain).order_by(Domain.name)
domains_result = await session.execute(domains_query)
domains = domains_result.scalars().all()
for domain in domains:
domains_stats.append(DomainStats(
id=domain.id,
name=domain.name,
color=domain.color,
mean=None,
elements_count=0
))
# Récupérer les compétences
competences_query = select(Competence).order_by(Competence.order_index)
competences_result = await session.execute(competences_query)
competences = competences_result.scalars().all()
for competence in competences:
competences_stats.append(CompetenceStats(
id=competence.id,
name=competence.name,
color=competence.color,
mean=None,
elements_count=0
))
return ClassDashboardStats(
class_id=class_id,
class_name=cls.name,
trimester=trimester,
students_count=len(students),
mean=mean,
median=median,
std_dev=std_dev,
min_score=min_score,
max_score=max_score,
assessments_total=len(assessments),
assessments_completed=assessments_completed,
assessments_in_progress=assessments_in_progress,
student_averages=student_averages,
histogram=histogram,
domains_stats=domains_stats,
competences_stats=competences_stats
)
@router.post("", response_model=ClassGroupRead, status_code=201)
async def create_class(
class_data: ClassGroupCreate,
session: AsyncSessionDep,
):
"""
Crée une nouvelle classe.
"""
# Vérifier l'unicité du nom
existing_query = select(ClassGroup).where(ClassGroup.name == class_data.name)
existing_result = await session.execute(existing_query)
if existing_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"Une classe avec le nom '{class_data.name}' existe déjà"
)
# Créer la nouvelle classe
new_class = ClassGroup(
name=class_data.name,
description=class_data.description,
year=class_data.year
)
session.add(new_class)
await session.commit()
await session.refresh(new_class)
return ClassGroupRead(
id=new_class.id,
name=new_class.name,
description=new_class.description,
year=new_class.year,
students_count=0
)
@router.put("/{class_id}", response_model=ClassGroupRead)
async def update_class(
class_id: int,
class_data: ClassGroupUpdate,
session: AsyncSessionDep,
):
"""
Modifie une classe existante.
"""
# Récupérer la classe
query = select(ClassGroup).where(ClassGroup.id == class_id)
result = await session.execute(query)
cls = result.scalar_one_or_none()
if not cls:
raise HTTPException(status_code=404, detail="Classe non trouvée")
# Vérifier l'unicité du nouveau nom si modifié
if class_data.name and class_data.name != cls.name:
existing_query = select(ClassGroup).where(
ClassGroup.name == class_data.name,
ClassGroup.id != class_id
)
existing_result = await session.execute(existing_query)
if existing_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"Une autre classe avec le nom '{class_data.name}' existe déjà"
)
# Appliquer les modifications
if class_data.name is not None:
cls.name = class_data.name
if class_data.description is not None:
cls.description = class_data.description
if class_data.year is not None:
cls.year = class_data.year
await session.commit()
await session.refresh(cls)
# Compter les étudiants
count_query = select(func.count(StudentEnrollment.id)).where(
StudentEnrollment.class_group_id == cls.id,
StudentEnrollment.departure_date.is_(None)
)
count_result = await session.execute(count_query)
students_count = count_result.scalar() or 0
return ClassGroupRead(
id=cls.id,
name=cls.name,
description=cls.description,
year=cls.year,
students_count=students_count
)
@router.delete("/{class_id}", status_code=204)
async def delete_class(
class_id: int,
session: AsyncSessionDep,
):
"""
Supprime une classe.
La suppression échoue si la classe contient des élèves ou des évaluations.
"""
# Récupérer la classe
query = select(ClassGroup).where(ClassGroup.id == class_id)
result = await session.execute(query)
cls = result.scalar_one_or_none()
if not cls:
raise HTTPException(status_code=404, detail="Classe non trouvée")
# Vérifier les dépendances : élèves inscrits
students_query = select(func.count(StudentEnrollment.id)).where(
StudentEnrollment.class_group_id == class_id
)
students_result = await session.execute(students_query)
students_count = students_result.scalar() or 0
# Vérifier les dépendances : évaluations
assessments_query = select(func.count(Assessment.id)).where(
Assessment.class_group_id == class_id
)
assessments_result = await session.execute(assessments_query)
assessments_count = assessments_result.scalar() or 0
if students_count > 0 or assessments_count > 0:
raise HTTPException(
status_code=400,
detail=f"Impossible de supprimer la classe '{cls.name}'. "
f"Elle contient {students_count} inscription(s) et {assessments_count} évaluation(s). "
f"Supprimez d'abord ces éléments."
)
# Supprimer la classe
await session.delete(cls)
await session.commit()
return None
def extract_name_parts(full_name: str) -> Tuple[str, str]:
"""
Extrait le nom et prénom depuis un nom complet.
Stratégie pour "AABIDA LAHDILI Fatima Zahra":
- Les mots en MAJUSCULES → nom de famille
- Les autres mots → prénoms
Args:
full_name: Nom complet de l'élève
Returns:
Tuple (nom_de_famille, prenoms)
"""
if not full_name or not full_name.strip():
return "", ""
# Nettoyer et séparer les mots
parts = full_name.strip().split()
if not parts:
return "", ""
# Séparer les mots en majuscules des autres
uppercase_words = []
other_words = []
for part in parts:
# Un mot est considéré en majuscules s'il contient au moins 2 caractères
# et que tous ses caractères alphabétiques sont en majuscules
if len(part) >= 2 and part.isupper():
uppercase_words.append(part)
else:
other_words.append(part)
# Le nom de famille = mots en majuscules
last_name = " ".join(uppercase_words) if uppercase_words else parts[0]
# Les prénoms = autres mots, ou le reste si pas de majuscules détectées
if uppercase_words and other_words:
first_name = " ".join(other_words)
elif len(parts) > 1:
# Fallback: premier mot = nom, reste = prénoms
last_name = parts[0]
first_name = " ".join(parts[1:])
else:
# Un seul mot : considérer comme prénom
first_name = parts[0]
last_name = ""
return last_name.strip(), first_name.strip()
@router.post("/{class_id}/import-csv", response_model=CSVImportResponse)
async def import_csv(
class_id: int,
session: AsyncSessionDep,
file: UploadFile = File(..., description="Fichier CSV à importer"),
enrollment_date: Optional[date] = Form(None, description="Date d'inscription (défaut: aujourd'hui)"),
skip_duplicates: bool = Form(True, description="Ignorer les doublons (True) ou échouer (False)"),
):
"""
Importe des élèves depuis un fichier CSV dans une classe.
Format CSV attendu:
- Séparateur: ;
- Première colonne: "NOM Prénoms" (ex: "DUPONT Marie Claire")
- Première ligne: headers (ignorée)
L'extraction du nom/prénom utilise la logique suivante:
- Mots en MAJUSCULES → nom de famille
- Autres mots → prénoms
"""
# Vérifier que la classe existe
class_query = select(ClassGroup).where(ClassGroup.id == class_id)
class_result = await session.execute(class_query)
cls = class_result.scalar_one_or_none()
if not cls:
raise HTTPException(status_code=404, detail="Classe non trouvée")
# Utiliser la date du jour si non spécifiée
if enrollment_date is None:
enrollment_date = date.today()
# Lire le contenu du fichier
try:
content = await file.read()
csv_content = content.decode('utf-8')
except UnicodeDecodeError:
# Essayer avec latin-1
try:
csv_content = content.decode('latin-1')
except Exception as e:
return CSVImportResponse(
success=False,
total_lines=0,
imported_count=0,
skipped_count=0,
error_count=1,
imported_students=[],
skipped_students=[],
errors=[ImportErrorInfo(line=0, error=f"Erreur de décodage du fichier: {str(e)}")],
message="Erreur de décodage du fichier"
)
# Parser le CSV
parsed_students = []
try:
csv_reader = csv.reader(StringIO(csv_content), delimiter=';')
# Ignorer la première ligne (headers)
headers = next(csv_reader, None)
if not headers:
return CSVImportResponse(
success=True,
total_lines=0,
imported_count=0,
skipped_count=0,
error_count=0,
imported_students=[],
skipped_students=[],
errors=[],
message="Fichier CSV vide"
)
for line_number, row in enumerate(csv_reader, start=2):
if not row or len(row) < 1:
continue
# Prendre la première colonne (nom complet)
full_name = row[0].strip().strip('"')
if not full_name or full_name.lower() in ['undefined', 'null', '']:
continue
# Extraire nom et prénom
last_name, first_name = extract_name_parts(full_name)
if not first_name and not last_name:
continue
parsed_students.append({
'first_name': first_name or "Prénom",
'last_name': last_name or "Nom",
'line_number': line_number,
'raw_name': full_name
})
except Exception as e:
return CSVImportResponse(
success=False,
total_lines=0,
imported_count=0,
skipped_count=0,
error_count=1,
imported_students=[],
skipped_students=[],
errors=[ImportErrorInfo(line=0, error=f"Erreur lors de l'analyse du fichier CSV: {str(e)}")],
message=f"Erreur lors de l'analyse du fichier CSV: {str(e)}"
)
if not parsed_students:
return CSVImportResponse(
success=True,
total_lines=0,
imported_count=0,
skipped_count=0,
error_count=0,
imported_students=[],
skipped_students=[],
errors=[],
message="Aucun élève trouvé dans le fichier CSV"
)
# Importer les élèves
imported_students = []
skipped_students = []
errors = []
for student_data in parsed_students:
try:
# Vérifier si l'élève existe déjà
existing_query = select(Student).where(
func.lower(Student.first_name) == student_data['first_name'].lower(),
func.lower(Student.last_name) == student_data['last_name'].lower()
)
existing_result = await session.execute(existing_query)
existing_student = existing_result.scalar_one_or_none()
if existing_student:
if skip_duplicates:
skipped_students.append(SkippedStudentInfo(
line=student_data['line_number'],
name=f"{student_data['first_name']} {student_data['last_name']}",
reason="Élève déjà existant"
))
continue
else:
errors.append(ImportErrorInfo(
line=student_data['line_number'],
error=f"Élève déjà existant: {student_data['first_name']} {student_data['last_name']}"
))
continue
# Créer le nouvel élève
new_student = Student(
first_name=student_data['first_name'],
last_name=student_data['last_name'],
email=None
)
session.add(new_student)
await session.flush() # Pour obtenir l'ID
# Créer l'inscription dans la classe
enrollment = StudentEnrollment(
student_id=new_student.id,
class_group_id=class_id,
enrollment_date=enrollment_date,
enrollment_reason="Import CSV"
)
session.add(enrollment)
imported_students.append(ImportedStudentInfo(
first_name=student_data['first_name'],
last_name=student_data['last_name'],
email=None,
line_number=student_data['line_number'],
raw_name=student_data['raw_name']
))
except Exception as e:
errors.append(ImportErrorInfo(
line=student_data['line_number'],
error=f"Erreur création élève {student_data['first_name']} {student_data['last_name']}: {str(e)}"
))
# Gérer les erreurs et le commit
if errors and not skip_duplicates:
await session.rollback()
return CSVImportResponse(
success=False,
total_lines=len(parsed_students),
imported_count=0,
skipped_count=len(skipped_students),
error_count=len(errors),
imported_students=[],
skipped_students=skipped_students,
errors=errors,
message=f"Import échoué: {len(errors)} erreur(s)"
)
await session.commit()
return CSVImportResponse(
success=True,
total_lines=len(parsed_students),
imported_count=len(imported_students),
skipped_count=len(skipped_students),
error_count=len(errors),
imported_students=imported_students,
skipped_students=skipped_students,
errors=errors,
message=f"Import réussi: {len(imported_students)} élève(s) importé(s), {len(skipped_students)} ignoré(s)"
)

View File

@@ -0,0 +1,978 @@
"""
Routes API pour la configuration (Competences, Domains, Scale).
"""
from typing import List
from fastapi import APIRouter, HTTPException
from sqlalchemy import select, func, delete
from api.dependencies import AsyncSessionDep
from infrastructure.database.models import (
AppConfig,
Competence,
Domain,
CompetenceScaleValue,
)
from schemas.config import (
CompetenceRead,
CompetenceList,
CompetenceCreate,
CompetenceUpdate,
DomainRead,
DomainList,
DomainCreate,
DomainUpdate,
ScaleValueRead,
ScaleRead,
ScaleValueCreate,
ScaleValueUpdate,
AppConfigRead,
AppConfigUpdate,
FullConfigRead,
ConfigResponse,
SMTPConfigRead,
SMTPConfigUpdate,
SMTPTestRequest,
SMTPTestResponse,
NotesGradientRead,
NotesGradientUpdate,
ScaleBatchUpdate,
DatabaseConfigRead,
DatabaseConfigUpdate,
DatabaseConfigResponse,
)
from infrastructure.external.email_service import EmailService
router = APIRouter(prefix="/config", tags=["Configuration"])
@router.get("/competences", response_model=CompetenceList)
async def get_competences(session: AsyncSessionDep):
"""
Récupère la liste de toutes les compétences.
"""
query = select(Competence).order_by(Competence.order_index)
result = await session.execute(query)
competences = result.scalars().all()
return CompetenceList(
competences=[
CompetenceRead(
id=c.id,
name=c.name,
color=c.color,
icon=c.icon,
order_index=c.order_index
)
for c in competences
]
)
@router.get("/domains", response_model=DomainList)
async def get_domains(session: AsyncSessionDep):
"""
Récupère la liste de tous les domaines.
"""
query = select(Domain).order_by(Domain.name)
result = await session.execute(query)
domains = result.scalars().all()
return DomainList(
domains=[
DomainRead(
id=d.id,
name=d.name,
color=d.color,
description=d.description
)
for d in domains
]
)
@router.get("/domains/search", response_model=DomainList)
async def search_domains(
q: str,
session: AsyncSessionDep,
):
"""
Recherche des domaines par nom (autocomplétion).
"""
query = select(Domain).where(
Domain.name.ilike(f"%{q}%")
).order_by(Domain.name).limit(10)
result = await session.execute(query)
domains = result.scalars().all()
return DomainList(
domains=[
DomainRead(
id=d.id,
name=d.name,
color=d.color,
description=d.description
)
for d in domains
]
)
@router.get("/scale", response_model=ScaleRead)
async def get_scale(session: AsyncSessionDep):
"""
Récupère l'échelle de compétences (valeurs 0, 1, 2, 3, ., d, etc.).
"""
query = select(CompetenceScaleValue).order_by(CompetenceScaleValue.value)
result = await session.execute(query)
scale_values = result.scalars().all()
return ScaleRead(
values=[
ScaleValueRead(
value=sv.value,
label=sv.label,
color=sv.color,
included_in_total=sv.included_in_total
)
for sv in scale_values
]
)
@router.get("", response_model=FullConfigRead)
async def get_full_config(session: AsyncSessionDep):
"""
Récupère la configuration complète de l'application.
"""
# App config
app_config_query = select(AppConfig)
app_config_result = await session.execute(app_config_query)
app_configs = app_config_result.scalars().all()
# Convertir en dictionnaire
config_dict = {c.key: c.value for c in app_configs}
app_config = AppConfigRead(
school_year=config_dict.get("school_year", "2024-2025"),
school_name=config_dict.get("school_name", ""),
default_grading_system=config_dict.get("default_grading_system", "notes")
)
# Competences
competences_query = select(Competence).order_by(Competence.order_index)
competences_result = await session.execute(competences_query)
competences = competences_result.scalars().all()
competences_list = [
CompetenceRead(
id=c.id,
name=c.name,
color=c.color,
icon=c.icon,
order_index=c.order_index
)
for c in competences
]
# Domains
domains_query = select(Domain).order_by(Domain.name)
domains_result = await session.execute(domains_query)
domains = domains_result.scalars().all()
domains_list = [
DomainRead(
id=d.id,
name=d.name,
color=d.color,
description=d.description
)
for d in domains
]
# Scale
scale_query = select(CompetenceScaleValue).order_by(CompetenceScaleValue.value)
scale_result = await session.execute(scale_query)
scale_values = scale_result.scalars().all()
scale = ScaleRead(
values=[
ScaleValueRead(
value=sv.value,
label=sv.label,
color=sv.color,
included_in_total=sv.included_in_total
)
for sv in scale_values
]
)
return FullConfigRead(
app_config=app_config,
competences=competences_list,
domains=domains_list,
scale=scale
)
# === CRUD Compétences ===
@router.post("/competences", response_model=CompetenceRead, status_code=201)
async def create_competence(
data: CompetenceCreate,
session: AsyncSessionDep,
):
"""
Crée une nouvelle compétence.
"""
# Vérifier l'unicité du nom
existing_query = select(Competence).where(Competence.name == data.name)
existing_result = await session.execute(existing_query)
if existing_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"Une compétence avec le nom '{data.name}' existe déjà"
)
# Déterminer l'index d'ordre
if data.order_index is None:
max_order_query = select(func.max(Competence.order_index))
max_order_result = await session.execute(max_order_query)
max_order = max_order_result.scalar() or 0
order_index = max_order + 1
else:
order_index = data.order_index
# Créer la compétence
competence = Competence(
name=data.name,
color=data.color,
icon=data.icon,
order_index=order_index
)
session.add(competence)
await session.commit()
await session.refresh(competence)
return CompetenceRead(
id=competence.id,
name=competence.name,
color=competence.color,
icon=competence.icon,
order_index=competence.order_index
)
@router.put("/competences/{competence_id}", response_model=CompetenceRead)
async def update_competence(
competence_id: int,
data: CompetenceUpdate,
session: AsyncSessionDep,
):
"""
Met à jour une compétence existante.
"""
query = select(Competence).where(Competence.id == competence_id)
result = await session.execute(query)
competence = result.scalar_one_or_none()
if not competence:
raise HTTPException(status_code=404, detail="Compétence non trouvée")
# Vérifier l'unicité du nouveau nom
if data.name and data.name != competence.name:
existing_query = select(Competence).where(
Competence.name == data.name,
Competence.id != competence_id
)
existing_result = await session.execute(existing_query)
if existing_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"Une autre compétence avec le nom '{data.name}' existe déjà"
)
# Appliquer les modifications
if data.name is not None:
competence.name = data.name
if data.color is not None:
competence.color = data.color
if data.icon is not None:
competence.icon = data.icon
if data.order_index is not None:
competence.order_index = data.order_index
await session.commit()
await session.refresh(competence)
return CompetenceRead(
id=competence.id,
name=competence.name,
color=competence.color,
icon=competence.icon,
order_index=competence.order_index
)
@router.delete("/competences/{competence_id}", response_model=ConfigResponse)
async def delete_competence(
competence_id: int,
session: AsyncSessionDep,
):
"""
Supprime une compétence.
"""
query = select(Competence).where(Competence.id == competence_id)
result = await session.execute(query)
competence = result.scalar_one_or_none()
if not competence:
raise HTTPException(status_code=404, detail="Compétence non trouvée")
name = competence.name
await session.delete(competence)
await session.commit()
return ConfigResponse(
success=True,
message=f"Compétence '{name}' supprimée"
)
# === CRUD Domaines ===
@router.post("/domains", response_model=DomainRead, status_code=201)
async def create_domain(
data: DomainCreate,
session: AsyncSessionDep,
):
"""
Crée un nouveau domaine.
"""
# Vérifier l'unicité du nom
existing_query = select(Domain).where(Domain.name == data.name)
existing_result = await session.execute(existing_query)
if existing_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"Un domaine avec le nom '{data.name}' existe déjà"
)
# Créer le domaine
domain = Domain(
name=data.name,
color=data.color,
description=data.description
)
session.add(domain)
await session.commit()
await session.refresh(domain)
return DomainRead(
id=domain.id,
name=domain.name,
color=domain.color,
description=domain.description
)
@router.put("/domains/{domain_id}", response_model=DomainRead)
async def update_domain(
domain_id: int,
data: DomainUpdate,
session: AsyncSessionDep,
):
"""
Met à jour un domaine existant.
"""
query = select(Domain).where(Domain.id == domain_id)
result = await session.execute(query)
domain = result.scalar_one_or_none()
if not domain:
raise HTTPException(status_code=404, detail="Domaine non trouvé")
# Vérifier l'unicité du nouveau nom
if data.name and data.name != domain.name:
existing_query = select(Domain).where(
Domain.name == data.name,
Domain.id != domain_id
)
existing_result = await session.execute(existing_query)
if existing_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"Un autre domaine avec le nom '{data.name}' existe déjà"
)
# Appliquer les modifications
if data.name is not None:
domain.name = data.name
if data.color is not None:
domain.color = data.color
if data.description is not None:
domain.description = data.description
await session.commit()
await session.refresh(domain)
return DomainRead(
id=domain.id,
name=domain.name,
color=domain.color,
description=domain.description
)
@router.delete("/domains/{domain_id}", response_model=ConfigResponse)
async def delete_domain(
domain_id: int,
session: AsyncSessionDep,
):
"""
Supprime un domaine.
"""
query = select(Domain).where(Domain.id == domain_id)
result = await session.execute(query)
domain = result.scalar_one_or_none()
if not domain:
raise HTTPException(status_code=404, detail="Domaine non trouvé")
name = domain.name
await session.delete(domain)
await session.commit()
return ConfigResponse(
success=True,
message=f"Domaine '{name}' supprimé"
)
# === CRUD Échelle de notation ===
@router.post("/scale", response_model=ScaleValueRead, status_code=201)
async def create_scale_value(
data: ScaleValueCreate,
session: AsyncSessionDep,
):
"""
Crée une nouvelle valeur d'échelle.
"""
# Vérifier l'unicité de la valeur
existing_query = select(CompetenceScaleValue).where(
CompetenceScaleValue.value == data.value
)
existing_result = await session.execute(existing_query)
if existing_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"Une valeur d'échelle '{data.value}' existe déjà"
)
# Créer la valeur
scale_value = CompetenceScaleValue(
value=data.value,
label=data.label,
color=data.color,
included_in_total=data.included_in_total
)
session.add(scale_value)
await session.commit()
await session.refresh(scale_value)
return ScaleValueRead(
value=scale_value.value,
label=scale_value.label,
color=scale_value.color,
included_in_total=scale_value.included_in_total
)
# Routes batch et reset AVANT les routes avec paramètres
@router.put("/scale/batch", response_model=ScaleRead)
async def update_scale_batch(
data: ScaleBatchUpdate,
session: AsyncSessionDep,
):
"""
Met à jour plusieurs valeurs d'échelle en une seule requête.
"""
for item in data.values:
query = select(CompetenceScaleValue).where(
CompetenceScaleValue.value == item.value
)
result = await session.execute(query)
sv = result.scalar_one_or_none()
if sv:
sv.label = item.label
sv.color = item.color
sv.included_in_total = item.included_in_total
else:
# Créer si n'existe pas
new_sv = CompetenceScaleValue(
value=item.value,
label=item.label,
color=item.color,
included_in_total=item.included_in_total
)
session.add(new_sv)
await session.commit()
return await get_scale(session)
@router.post("/scale/reset", response_model=ScaleRead)
async def reset_scale(session: AsyncSessionDep):
"""
Réinitialise l'échelle aux valeurs par défaut.
"""
# Supprimer toutes les valeurs existantes
await session.execute(
delete(CompetenceScaleValue)
)
# Valeurs par défaut
default_values = [
("0", "Non acquis", "#dc2626", True),
("1", "En cours d'acquisition", "#ea580c", True),
("2", "Acquis", "#059669", True),
("3", "Expert", "#2563eb", True),
(".", "Non évalué", "#6b7280", True),
]
for value, label, color, included in default_values:
sv = CompetenceScaleValue(
value=value,
label=label,
color=color,
included_in_total=included
)
session.add(sv)
await session.commit()
return await get_scale(session)
@router.put("/scale/{scale_value}", response_model=ScaleValueRead)
async def update_scale_value(
scale_value: str,
data: ScaleValueUpdate,
session: AsyncSessionDep,
):
"""
Met à jour une valeur d'échelle existante.
"""
query = select(CompetenceScaleValue).where(
CompetenceScaleValue.value == scale_value
)
result = await session.execute(query)
sv = result.scalar_one_or_none()
if not sv:
raise HTTPException(status_code=404, detail="Valeur d'échelle non trouvée")
# Vérifier l'unicité de la nouvelle valeur
if data.value and data.value != sv.value:
existing_query = select(CompetenceScaleValue).where(
CompetenceScaleValue.value == data.value
)
existing_result = await session.execute(existing_query)
if existing_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"Une autre valeur d'échelle '{data.value}' existe déjà"
)
# Appliquer les modifications
if data.value is not None:
sv.value = data.value
if data.label is not None:
sv.label = data.label
if data.color is not None:
sv.color = data.color
if data.included_in_total is not None:
sv.included_in_total = data.included_in_total
await session.commit()
await session.refresh(sv)
return ScaleValueRead(
value=sv.value,
label=sv.label,
color=sv.color,
included_in_total=sv.included_in_total
)
@router.delete("/scale/{scale_value}", response_model=ConfigResponse)
async def delete_scale_value(
scale_value: str,
session: AsyncSessionDep,
):
"""
Supprime une valeur d'échelle.
Note: Les valeurs de base (0, 1, 2, 3, .) ne peuvent pas être supprimées.
"""
# Protection des valeurs de base
base_values = ['0', '1', '2', '3', '.']
if scale_value in base_values:
raise HTTPException(
status_code=400,
detail=f"La valeur '{scale_value}' est une valeur de base et ne peut pas être supprimée"
)
query = select(CompetenceScaleValue).where(
CompetenceScaleValue.value == scale_value
)
result = await session.execute(query)
sv = result.scalar_one_or_none()
if not sv:
raise HTTPException(status_code=404, detail="Valeur d'échelle non trouvée")
value = sv.value
await session.delete(sv)
await session.commit()
return ConfigResponse(
success=True,
message=f"Valeur d'échelle '{value}' supprimée"
)
# === Configuration générale ===
@router.put("", response_model=AppConfigRead)
async def update_app_config(
data: AppConfigUpdate,
session: AsyncSessionDep,
):
"""
Met à jour la configuration générale de l'application.
"""
updates = []
if data.school_year is not None:
updates.append(("school_year", data.school_year))
if data.school_name is not None:
updates.append(("school_name", data.school_name))
if data.default_grading_system is not None:
updates.append(("default_grading_system", data.default_grading_system))
for key, value in updates:
# Chercher la config existante
query = select(AppConfig).where(AppConfig.key == key)
result = await session.execute(query)
config = result.scalar_one_or_none()
if config:
config.value = value
else:
# Créer si n'existe pas
new_config = AppConfig(key=key, value=value)
session.add(new_config)
await session.commit()
# Récupérer la configuration mise à jour
config_query = select(AppConfig)
config_result = await session.execute(config_query)
configs = config_result.scalars().all()
config_dict = {c.key: c.value for c in configs}
return AppConfigRead(
school_year=config_dict.get("school_year", "2024-2025"),
school_name=config_dict.get("school_name", ""),
default_grading_system=config_dict.get("default_grading_system", "notes")
)
# === Configuration SMTP Email ===
@router.get("/smtp", response_model=SMTPConfigRead)
async def get_smtp_config(session: AsyncSessionDep):
"""
Récupère la configuration SMTP pour l'envoi d'emails.
Note: Le mot de passe n'est jamais retourné pour des raisons de sécurité.
"""
# Récupérer toutes les configs email
query = select(AppConfig).where(AppConfig.key.like("email.%"))
result = await session.execute(query)
configs = result.scalars().all()
config_dict = {c.key: c.value for c in configs}
# Créer le service pour vérifier si configuré
email_service = EmailService.from_database_config(config_dict)
return SMTPConfigRead(
host=config_dict.get("email.smtp_host", ""),
port=int(config_dict.get("email.smtp_port", 587)),
username=config_dict.get("email.username", ""),
use_tls=str(config_dict.get("email.use_tls", "true")).lower() == "true",
from_name=config_dict.get("email.from_name", "Notytex"),
from_address=config_dict.get("email.from_address", ""),
is_configured=email_service.is_configured()
)
@router.put("/smtp", response_model=SMTPConfigRead)
async def update_smtp_config(
data: SMTPConfigUpdate,
session: AsyncSessionDep,
):
"""
Met à jour la configuration SMTP.
Tous les champs sont optionnels - seuls ceux fournis seront mis à jour.
"""
updates = []
if data.host is not None:
updates.append(("email.smtp_host", data.host))
if data.port is not None:
updates.append(("email.smtp_port", str(data.port)))
if data.username is not None:
updates.append(("email.username", data.username))
if data.password is not None:
updates.append(("email.password", data.password))
if data.use_tls is not None:
updates.append(("email.use_tls", "true" if data.use_tls else "false"))
if data.from_name is not None:
updates.append(("email.from_name", data.from_name))
if data.from_address is not None:
updates.append(("email.from_address", data.from_address))
for key, value in updates:
# Chercher la config existante
query = select(AppConfig).where(AppConfig.key == key)
result = await session.execute(query)
config = result.scalar_one_or_none()
if config:
config.value = value
else:
# Créer si n'existe pas
new_config = AppConfig(key=key, value=value)
session.add(new_config)
await session.commit()
# Retourner la config mise à jour
return await get_smtp_config(session)
@router.post("/smtp/test", response_model=SMTPTestResponse)
async def test_smtp_config(
data: SMTPTestRequest,
session: AsyncSessionDep,
):
"""
Teste la configuration SMTP en envoyant un email de test.
"""
# Récupérer la config
query = select(AppConfig).where(AppConfig.key.like("email.%"))
result = await session.execute(query)
configs = result.scalars().all()
config_dict = {c.key: c.value for c in configs}
# Créer le service et envoyer le test
email_service = EmailService.from_database_config(config_dict)
if not email_service.is_configured():
return SMTPTestResponse(
success=False,
message="Configuration SMTP incomplète. Vérifiez les paramètres."
)
result = email_service.send_test_email(data.test_email)
return SMTPTestResponse(
success=result.get("success", False),
message=result.get("message", result.get("error", "Erreur inconnue"))
)
# === Configuration dégradé des notes ===
@router.get("/notes-gradient", response_model=NotesGradientRead)
async def get_notes_gradient(session: AsyncSessionDep):
"""
Récupère la configuration du dégradé de couleurs pour les notes.
"""
query = select(AppConfig).where(AppConfig.key.like("grading.notes_gradient.%"))
result = await session.execute(query)
configs = result.scalars().all()
config_dict = {c.key: c.value for c in configs}
return NotesGradientRead(
min_color=config_dict.get("grading.notes_gradient.min_color", "#dc2626"),
max_color=config_dict.get("grading.notes_gradient.max_color", "#059669"),
enabled=str(config_dict.get("grading.notes_gradient.enabled", "false")).lower() == "true"
)
@router.put("/notes-gradient", response_model=NotesGradientRead)
async def update_notes_gradient(
data: NotesGradientUpdate,
session: AsyncSessionDep,
):
"""
Met à jour la configuration du dégradé de couleurs pour les notes.
"""
updates = []
if data.min_color is not None:
updates.append(("grading.notes_gradient.min_color", data.min_color))
if data.max_color is not None:
updates.append(("grading.notes_gradient.max_color", data.max_color))
if data.enabled is not None:
updates.append(("grading.notes_gradient.enabled", "true" if data.enabled else "false"))
for key, value in updates:
query = select(AppConfig).where(AppConfig.key == key)
result = await session.execute(query)
config = result.scalar_one_or_none()
if config:
config.value = value
else:
new_config = AppConfig(key=key, value=value)
session.add(new_config)
await session.commit()
return await get_notes_gradient(session)
# === Configuration de la base de données ===
def _format_size(size_bytes: int) -> str:
"""Formate une taille en bytes en format lisible."""
size = float(size_bytes)
for unit in ['B', 'KB', 'MB', 'GB']:
if size < 1024:
return f"{size:.1f} {unit}"
size /= 1024
return f"{size:.1f} TB"
@router.get("/database", response_model=DatabaseConfigRead)
async def get_database_config():
"""
Récupère la configuration actuelle de la base de données.
Retourne le chemin configuré, le chemin résolu, et des informations
sur le fichier (existence, taille).
"""
from core.config import settings
from pathlib import Path
resolved_path = settings.database_path_resolved
exists = resolved_path.exists()
size_bytes = None
size_human = None
if exists:
size_bytes = resolved_path.stat().st_size
size_human = _format_size(size_bytes)
return DatabaseConfigRead(
path=settings.database_path_configured,
resolved_path=str(resolved_path),
exists=exists,
size_bytes=size_bytes,
size_human=size_human
)
@router.put("/database", response_model=DatabaseConfigResponse)
async def update_database_config(
data: DatabaseConfigUpdate,
):
"""
Met à jour le chemin de la base de données.
IMPORTANT: Cette modification met à jour le fichier .env et nécessite
un redémarrage de l'application pour prendre effet.
Le chemin peut être:
- Relatif au répertoire backend (ex: "data/school.db")
- Absolu (ex: "/var/data/notytex/school.db")
"""
from pathlib import Path
import os
# Valider que le chemin est valide
new_path = Path(data.path)
# Si relatif, vérifier que le répertoire parent existe ou peut être créé
if not new_path.is_absolute():
resolved = Path(__file__).parent.parent.parent / data.path
else:
resolved = new_path
# Vérifier que le répertoire parent existe
parent_dir = resolved.parent
if not parent_dir.exists():
try:
parent_dir.mkdir(parents=True, exist_ok=True)
except Exception as e:
raise HTTPException(
status_code=400,
detail=f"Impossible de créer le répertoire parent: {str(e)}"
)
# Trouver le fichier .env
env_file = Path(__file__).parent.parent.parent / ".env"
# Lire le contenu actuel ou créer un nouveau fichier
env_content = ""
if env_file.exists():
env_content = env_file.read_text()
# Mettre à jour ou ajouter DATABASE_PATH
import re
if "DATABASE_PATH=" in env_content:
# Remplacer la valeur existante
env_content = re.sub(
r'^DATABASE_PATH=.*$',
f'DATABASE_PATH={data.path}',
env_content,
flags=re.MULTILINE
)
else:
# Ajouter la nouvelle variable
if env_content and not env_content.endswith('\n'):
env_content += '\n'
env_content += f'DATABASE_PATH={data.path}\n'
# Écrire le fichier .env
try:
env_file.write_text(env_content)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Erreur lors de l'écriture du fichier .env: {str(e)}"
)
return DatabaseConfigResponse(
success=True,
message=f"Chemin de base de données mis à jour. Redémarrez l'application pour appliquer les changements.",
requires_restart=True,
new_path=data.path
)

View File

@@ -0,0 +1,439 @@
"""
Routes API pour le conseil de classe (préparation et appréciations).
"""
from typing import Optional
from fastapi import APIRouter, HTTPException, Query
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
from api.dependencies import AsyncSessionDep
from infrastructure.database.models import (
ClassGroup,
Student,
StudentEnrollment,
Assessment,
Exercise,
GradingElement,
Grade,
CouncilAppreciation,
)
from domain.services.grading_calculator import GradingCalculator
from schemas.council import (
CouncilPreparationRead,
StudentTrimesterSummaryRead,
ClassStatisticsRead,
AppreciationStatsRead,
PerformanceDistribution,
AssessmentGradeInfo,
AppreciationCreate,
AppreciationRead,
AppreciationResponse,
)
router = APIRouter(prefix="/council", tags=["Conseil de Classe"])
async def get_current_students_in_class(session, class_id: int):
"""Récupère les élèves actuellement inscrits dans une classe."""
query = (
select(Student)
.join(StudentEnrollment, Student.id == StudentEnrollment.student_id)
.where(
StudentEnrollment.class_group_id == class_id,
StudentEnrollment.departure_date.is_(None)
)
.order_by(Student.last_name, Student.first_name)
)
result = await session.execute(query)
return result.scalars().all()
async def calculate_student_assessment_score(session, student_id: int, assessment_id: int) -> Optional[float]:
"""Calcule le score d'un élève pour une évaluation, ramené sur 20."""
# Récupérer les notes de l'élève pour cette évaluation
query = (
select(Grade, GradingElement)
.join(GradingElement, Grade.grading_element_id == GradingElement.id)
.join(Exercise, GradingElement.exercise_id == Exercise.id)
.where(
Exercise.assessment_id == assessment_id,
Grade.student_id == student_id
)
)
result = await session.execute(query)
grades_data = result.all()
if not grades_data:
return None
calculator = GradingCalculator()
total_score = 0.0
total_max_points = 0.0
for grade, element in grades_data:
if grade.value:
score = calculator.calculate_score(
grade.value, element.grading_type, element.max_points
)
if score is not None and calculator.is_counted_in_total(grade.value):
total_score += score
total_max_points += element.max_points
if total_max_points > 0:
return round(total_score / total_max_points * 20, 2)
return None
async def get_student_trimester_average(session, student_id: int, class_id: int, trimester: int) -> Optional[float]:
"""Calcule la moyenne pondérée d'un élève pour un trimestre."""
# Récupérer les évaluations du trimestre avec leurs coefficients
assessments_query = select(Assessment).where(
Assessment.class_group_id == class_id,
Assessment.trimester == trimester
)
result = await session.execute(assessments_query)
assessments = result.scalars().all()
if not assessments:
return None
weighted_sum = 0.0
total_coefficient = 0.0
for assessment in assessments:
score = await calculate_student_assessment_score(session, student_id, assessment.id)
if score is not None:
weighted_sum += score * assessment.coefficient
total_coefficient += assessment.coefficient
if total_coefficient > 0:
return round(weighted_sum / total_coefficient, 2)
return None
def determine_performance_status(average: Optional[float]) -> str:
"""Détermine le statut de performance d'un élève."""
if average is None:
return "no_data"
if average >= 16:
return "excellent"
elif average >= 14:
return "good"
elif average >= 10:
return "average"
else:
return "struggling"
@router.get("/classes/{class_id}", response_model=CouncilPreparationRead)
async def get_council_preparation(
class_id: int,
trimester: int,
session: AsyncSessionDep,
):
"""
Récupère les données complètes pour la préparation du conseil de classe.
Inclut:
- Résumé de chaque élève (moyenne, nombre d'évaluations, performances)
- Statistiques de classe (moyenne, médiane, écart-type)
- État des appréciations (complétées/en cours)
"""
# Vérifier que la classe existe
class_query = select(ClassGroup).where(ClassGroup.id == class_id)
class_result = await session.execute(class_query)
cls = class_result.scalar_one_or_none()
if not cls:
raise HTTPException(status_code=404, detail="Classe non trouvée")
# Récupérer les élèves actuellement dans la classe
students = await get_current_students_in_class(session, class_id)
# Récupérer les évaluations du trimestre
assessments_query = select(Assessment).where(
Assessment.class_group_id == class_id,
Assessment.trimester == trimester
).order_by(Assessment.date)
assessments_result = await session.execute(assessments_query)
assessments = assessments_result.scalars().all()
# Récupérer les appréciations existantes
appreciations_query = select(CouncilAppreciation).where(
CouncilAppreciation.class_group_id == class_id,
CouncilAppreciation.trimester == trimester
)
appreciations_result = await session.execute(appreciations_query)
appreciations = {a.student_id: a for a in appreciations_result.scalars().all()}
# Générer les résumés par élève
student_summaries = []
all_averages = []
for student in students:
# Calculer les notes par évaluation
grades_by_assessment = {}
assessment_count = 0
for assessment in assessments:
score = await calculate_student_assessment_score(session, student.id, assessment.id)
if score is not None:
assessment_count += 1
grades_by_assessment[assessment.id] = AssessmentGradeInfo(
id=assessment.id,
title=assessment.title,
date=assessment.date,
score=score,
max_score=20.0,
coefficient=assessment.coefficient
)
# Calculer la moyenne générale
overall_average = await get_student_trimester_average(session, student.id, class_id, trimester)
if overall_average is not None:
all_averages.append(overall_average)
# Récupérer l'appréciation
appreciation = appreciations.get(student.id)
student_summaries.append(StudentTrimesterSummaryRead(
student_id=student.id,
first_name=student.first_name,
last_name=student.last_name,
full_name=f"{student.first_name} {student.last_name}",
overall_average=overall_average,
assessment_count=assessment_count,
grades_by_assessment=grades_by_assessment,
performance_status=determine_performance_status(overall_average),
has_appreciation=appreciation is not None and (
appreciation.general_appreciation or
appreciation.strengths or
appreciation.areas_for_improvement
),
appreciation_status=appreciation.status if appreciation else None
))
# Calculer les statistiques de classe
if all_averages:
mean = round(sum(all_averages) / len(all_averages), 2)
sorted_averages = sorted(all_averages)
n = len(sorted_averages)
if n % 2 == 0:
median = (sorted_averages[n//2 - 1] + sorted_averages[n//2]) / 2
else:
median = sorted_averages[n//2]
median = round(median, 2)
min_avg = min(all_averages)
max_avg = max(all_averages)
# Écart-type
variance = sum((x - mean) ** 2 for x in all_averages) / len(all_averages)
std_dev = round(variance ** 0.5, 2)
else:
mean = median = min_avg = max_avg = std_dev = None
# Distribution des performances
performance_dist = {
"excellent": 0,
"good": 0,
"average": 0,
"struggling": 0,
"no_data": 0
}
for summary in student_summaries:
performance_dist[summary.performance_status] += 1
class_statistics = ClassStatisticsRead(
mean=mean,
median=median,
min=min_avg,
max=max_avg,
std_dev=std_dev,
performance_distribution=PerformanceDistribution(**performance_dist),
student_count_with_data=len(all_averages),
total_students=len(students)
)
# Statistiques des appréciations
draft_count = sum(1 for a in appreciations.values() if a.status == "draft")
finalized_count = sum(1 for a in appreciations.values() if a.status == "finalized")
completed_count = sum(1 for s in student_summaries if s.has_appreciation)
appreciation_stats = AppreciationStatsRead(
total_students=len(students),
completed_appreciations=completed_count,
draft_appreciations=draft_count,
finalized_appreciations=finalized_count,
completion_percentage=round(completed_count / len(students) * 100, 1) if students else 0
)
return CouncilPreparationRead(
class_group_id=class_id,
class_name=cls.name,
trimester=trimester,
student_summaries=student_summaries,
class_statistics=class_statistics,
appreciation_stats=appreciation_stats,
total_students=len(students),
completed_appreciations=completed_count
)
@router.get("/classes/{class_id}/appreciations/{student_id}", response_model=AppreciationRead)
async def get_appreciation(
class_id: int,
student_id: int,
trimester: int,
session: AsyncSessionDep,
):
"""
Récupère l'appréciation d'un élève pour un trimestre.
"""
query = select(CouncilAppreciation).where(
CouncilAppreciation.class_group_id == class_id,
CouncilAppreciation.student_id == student_id,
CouncilAppreciation.trimester == trimester
)
result = await session.execute(query)
appreciation = result.scalar_one_or_none()
if not appreciation:
raise HTTPException(status_code=404, detail="Appréciation non trouvée")
return AppreciationRead(
id=appreciation.id,
student_id=appreciation.student_id,
class_group_id=appreciation.class_group_id,
trimester=appreciation.trimester,
general_appreciation=appreciation.general_appreciation,
strengths=appreciation.strengths,
areas_for_improvement=appreciation.areas_for_improvement,
status=appreciation.status,
last_modified=appreciation.last_modified,
created_at=appreciation.created_at
)
@router.post("/classes/{class_id}/appreciations/{student_id}", response_model=AppreciationResponse)
async def create_or_update_appreciation(
class_id: int,
student_id: int,
trimester: int,
data: AppreciationCreate,
session: AsyncSessionDep,
):
"""
Crée ou met à jour l'appréciation d'un élève pour un trimestre.
"""
# Vérifier que la classe existe
class_query = select(ClassGroup).where(ClassGroup.id == class_id)
class_result = await session.execute(class_query)
cls = class_result.scalar_one_or_none()
if not cls:
raise HTTPException(status_code=404, detail="Classe non trouvée")
# Vérifier que l'élève existe
student_query = select(Student).where(Student.id == student_id)
student_result = await session.execute(student_query)
student = student_result.scalar_one_or_none()
if not student:
raise HTTPException(status_code=404, detail="Élève non trouvé")
# Chercher une appréciation existante
existing_query = select(CouncilAppreciation).where(
CouncilAppreciation.class_group_id == class_id,
CouncilAppreciation.student_id == student_id,
CouncilAppreciation.trimester == trimester
)
existing_result = await session.execute(existing_query)
appreciation = existing_result.scalar_one_or_none()
if appreciation:
# Mettre à jour l'existante
appreciation.general_appreciation = data.general_appreciation
appreciation.strengths = data.strengths
appreciation.areas_for_improvement = data.areas_for_improvement
appreciation.status = data.status
message = "Appréciation mise à jour"
else:
# Créer une nouvelle
appreciation = CouncilAppreciation(
student_id=student_id,
class_group_id=class_id,
trimester=trimester,
general_appreciation=data.general_appreciation,
strengths=data.strengths,
areas_for_improvement=data.areas_for_improvement,
status=data.status
)
session.add(appreciation)
message = "Appréciation créée"
await session.commit()
await session.refresh(appreciation)
return AppreciationResponse(
success=True,
message=message,
appreciation=AppreciationRead(
id=appreciation.id,
student_id=appreciation.student_id,
class_group_id=appreciation.class_group_id,
trimester=appreciation.trimester,
general_appreciation=appreciation.general_appreciation,
strengths=appreciation.strengths,
areas_for_improvement=appreciation.areas_for_improvement,
status=appreciation.status,
last_modified=appreciation.last_modified,
created_at=appreciation.created_at
)
)
@router.put("/classes/{class_id}/appreciations/{student_id}/finalize", response_model=AppreciationResponse)
async def finalize_appreciation(
class_id: int,
student_id: int,
trimester: int,
session: AsyncSessionDep,
):
"""
Finalise une appréciation (change le statut à 'finalized').
"""
query = select(CouncilAppreciation).where(
CouncilAppreciation.class_group_id == class_id,
CouncilAppreciation.student_id == student_id,
CouncilAppreciation.trimester == trimester
)
result = await session.execute(query)
appreciation = result.scalar_one_or_none()
if not appreciation:
raise HTTPException(status_code=404, detail="Appréciation non trouvée")
appreciation.status = "finalized"
await session.commit()
await session.refresh(appreciation)
return AppreciationResponse(
success=True,
message="Appréciation finalisée",
appreciation=AppreciationRead(
id=appreciation.id,
student_id=appreciation.student_id,
class_group_id=appreciation.class_group_id,
trimester=appreciation.trimester,
general_appreciation=appreciation.general_appreciation,
strengths=appreciation.strengths,
areas_for_improvement=appreciation.areas_for_improvement,
status=appreciation.status,
last_modified=appreciation.last_modified,
created_at=appreciation.created_at
)
)

View File

@@ -0,0 +1,512 @@
"""
Routes API pour les étudiants (Student).
"""
from typing import Optional
from datetime import date
from fastapi import APIRouter, HTTPException, Query
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
from api.dependencies import AsyncSessionDep
from infrastructure.database.models import (
Student,
StudentEnrollment,
ClassGroup,
)
from schemas.student import (
StudentRead,
StudentWithClass,
StudentDetail,
StudentList,
EnrollmentRead,
StudentCreate,
StudentUpdate,
EnrollRequest,
TransferRequest,
DepartureRequest,
EnrollmentResponse,
TransferResponse,
DepartureResponse,
)
router = APIRouter(prefix="/students", tags=["Students"])
@router.get("", response_model=StudentList)
async def get_students(
session: AsyncSessionDep,
search: Optional[str] = Query(None, description="Recherche par nom ou prénom"),
class_id: Optional[int] = Query(None, description="Filtrer par classe actuelle"),
):
"""
Récupère la liste de tous les étudiants.
"""
query = select(Student).options(
selectinload(Student.enrollments).selectinload(StudentEnrollment.class_group)
)
if search:
search_term = f"%{search}%"
query = query.where(
(Student.last_name.ilike(search_term)) |
(Student.first_name.ilike(search_term))
)
query = query.order_by(Student.last_name, Student.first_name)
result = await session.execute(query)
students = result.scalars().all()
students_list = []
for student in students:
# Trouver l'inscription active
current_enrollment = None
for enrollment in student.enrollments:
if enrollment.departure_date is None:
current_enrollment = enrollment
break
# Filtrer par classe si demandé
if class_id and (not current_enrollment or current_enrollment.class_group_id != class_id):
continue
students_list.append(
StudentWithClass(
id=student.id,
last_name=student.last_name,
first_name=student.first_name,
email=student.email,
full_name=f"{student.first_name} {student.last_name}",
current_class_id=current_enrollment.class_group_id if current_enrollment else None,
current_class_name=current_enrollment.class_group.name if current_enrollment else None
)
)
return StudentList(
students=students_list,
total=len(students_list)
)
@router.get("/{student_id}", response_model=StudentDetail)
async def get_student(
student_id: int,
session: AsyncSessionDep,
):
"""
Récupère les détails d'un étudiant spécifique avec son historique d'inscriptions.
"""
query = (
select(Student)
.options(
selectinload(Student.enrollments).selectinload(StudentEnrollment.class_group)
)
.where(Student.id == student_id)
)
result = await session.execute(query)
student = result.scalar_one_or_none()
if not student:
raise HTTPException(status_code=404, detail="Étudiant non trouvé")
# Trouver l'inscription active
current_enrollment = None
enrollments_list = []
for enrollment in student.enrollments:
if enrollment.departure_date is None:
current_enrollment = enrollment
enrollments_list.append(
EnrollmentRead(
id=enrollment.id,
student_id=enrollment.student_id,
class_group_id=enrollment.class_group_id,
class_name=enrollment.class_group.name,
enrollment_date=enrollment.enrollment_date,
departure_date=enrollment.departure_date,
enrollment_reason=enrollment.enrollment_reason,
departure_reason=enrollment.departure_reason,
is_active=enrollment.departure_date is None,
created_at=enrollment.created_at,
updated_at=enrollment.updated_at
)
)
# Trier les inscriptions par date décroissante
enrollments_list.sort(key=lambda e: e.enrollment_date, reverse=True)
return StudentDetail(
id=student.id,
last_name=student.last_name,
first_name=student.first_name,
email=student.email,
full_name=f"{student.first_name} {student.last_name}",
current_class_id=current_enrollment.class_group_id if current_enrollment else None,
current_class_name=current_enrollment.class_group.name if current_enrollment else None,
enrollments=enrollments_list
)
@router.post("", response_model=StudentWithClass, status_code=201)
async def create_student(
student_data: StudentCreate,
session: AsyncSessionDep,
):
"""
Crée un nouvel étudiant et l'inscrit optionnellement dans une classe.
"""
# Vérifier l'unicité de l'email si fourni
if student_data.email:
existing_query = select(Student).where(Student.email == student_data.email)
existing_result = await session.execute(existing_query)
if existing_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"Un élève avec l'email '{student_data.email}' existe déjà"
)
# Créer l'étudiant
new_student = Student(
last_name=student_data.last_name,
first_name=student_data.first_name,
email=student_data.email
)
session.add(new_student)
await session.flush() # Pour obtenir l'ID
current_class_id = None
current_class_name = None
# Créer l'inscription si class_group_id est fourni
if student_data.class_group_id:
# Vérifier que la classe existe
class_query = select(ClassGroup).where(ClassGroup.id == student_data.class_group_id)
class_result = await session.execute(class_query)
class_group = class_result.scalar_one_or_none()
if not class_group:
raise HTTPException(status_code=404, detail="Classe non trouvée")
enrollment = StudentEnrollment(
student_id=new_student.id,
class_group_id=student_data.class_group_id,
enrollment_date=student_data.enrollment_date or date.today(),
enrollment_reason=student_data.enrollment_reason
)
session.add(enrollment)
current_class_id = class_group.id
current_class_name = class_group.name
await session.commit()
return StudentWithClass(
id=new_student.id,
last_name=new_student.last_name,
first_name=new_student.first_name,
email=new_student.email,
full_name=f"{new_student.first_name} {new_student.last_name}",
current_class_id=current_class_id,
current_class_name=current_class_name
)
@router.put("/{student_id}", response_model=StudentWithClass)
async def update_student(
student_id: int,
student_data: StudentUpdate,
session: AsyncSessionDep,
):
"""
Modifie les informations d'un étudiant.
"""
# Récupérer l'étudiant
query = (
select(Student)
.options(
selectinload(Student.enrollments).selectinload(StudentEnrollment.class_group)
)
.where(Student.id == student_id)
)
result = await session.execute(query)
student = result.scalar_one_or_none()
if not student:
raise HTTPException(status_code=404, detail="Étudiant non trouvé")
# Vérifier l'unicité du nouvel email si modifié
if student_data.email and student_data.email != student.email:
existing_query = select(Student).where(
Student.email == student_data.email,
Student.id != student_id
)
existing_result = await session.execute(existing_query)
if existing_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"Un autre élève avec l'email '{student_data.email}' existe déjà"
)
# Appliquer les modifications
if student_data.last_name is not None:
student.last_name = student_data.last_name
if student_data.first_name is not None:
student.first_name = student_data.first_name
if student_data.email is not None:
student.email = student_data.email if student_data.email else None
await session.commit()
await session.refresh(student)
# Trouver la classe actuelle
current_enrollment = None
for enrollment in student.enrollments:
if enrollment.departure_date is None:
current_enrollment = enrollment
break
return StudentWithClass(
id=student.id,
last_name=student.last_name,
first_name=student.first_name,
email=student.email,
full_name=f"{student.first_name} {student.last_name}",
current_class_id=current_enrollment.class_group_id if current_enrollment else None,
current_class_name=current_enrollment.class_group.name if current_enrollment else None
)
@router.delete("/{student_id}", status_code=204)
async def delete_student(
student_id: int,
session: AsyncSessionDep,
):
"""
Supprime un étudiant et toutes ses inscriptions.
"""
# Récupérer l'étudiant
query = select(Student).where(Student.id == student_id)
result = await session.execute(query)
student = result.scalar_one_or_none()
if not student:
raise HTTPException(status_code=404, detail="Étudiant non trouvé")
# Supprimer l'étudiant (les inscriptions seront supprimées en cascade)
await session.delete(student)
await session.commit()
return None
@router.post("/enroll", response_model=EnrollmentResponse, status_code=201)
async def enroll_student(
enroll_data: EnrollRequest,
session: AsyncSessionDep,
):
"""
Inscrit un élève dans une classe.
Peut créer un nouvel élève si student_id n'est pas fourni.
"""
# Vérifier que la classe existe
class_query = select(ClassGroup).where(ClassGroup.id == enroll_data.class_group_id)
class_result = await session.execute(class_query)
class_group = class_result.scalar_one_or_none()
if not class_group:
raise HTTPException(status_code=404, detail="Classe non trouvée")
is_new_student = False
if enroll_data.student_id:
# Élève existant
student_query = select(Student).where(Student.id == enroll_data.student_id)
student_result = await session.execute(student_query)
student = student_result.scalar_one_or_none()
if not student:
raise HTTPException(status_code=404, detail="Élève non trouvé")
# Vérifier qu'il n'a pas déjà une inscription active
active_query = select(StudentEnrollment).where(
StudentEnrollment.student_id == enroll_data.student_id,
StudentEnrollment.departure_date.is_(None)
)
active_result = await session.execute(active_query)
if active_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"L'élève {student.first_name} {student.last_name} est déjà inscrit dans une classe"
)
else:
# Nouvel élève
if not enroll_data.first_name or not enroll_data.last_name:
raise HTTPException(
status_code=400,
detail="Le prénom et le nom sont obligatoires pour créer un nouvel élève"
)
# Vérifier l'unicité de l'email si fourni
if enroll_data.email:
existing_query = select(Student).where(Student.email == enroll_data.email)
existing_result = await session.execute(existing_query)
if existing_result.scalar_one_or_none():
raise HTTPException(
status_code=400,
detail=f"Un élève avec l'email '{enroll_data.email}' existe déjà"
)
# Créer le nouvel élève
student = Student(
first_name=enroll_data.first_name,
last_name=enroll_data.last_name,
email=enroll_data.email
)
session.add(student)
await session.flush()
is_new_student = True
# Créer l'inscription
enrollment = StudentEnrollment(
student_id=student.id,
class_group_id=enroll_data.class_group_id,
enrollment_date=enroll_data.enrollment_date,
enrollment_reason=enroll_data.enrollment_reason
)
session.add(enrollment)
await session.commit()
await session.refresh(enrollment)
return EnrollmentResponse(
enrollment_id=enrollment.id,
student_id=student.id,
student_name=f"{student.first_name} {student.last_name}",
class_name=class_group.name,
message=f"Élève {student.first_name} {student.last_name} inscrit en {class_group.name}",
is_new_student=is_new_student
)
@router.post("/transfer", response_model=TransferResponse)
async def transfer_student(
transfer_data: TransferRequest,
session: AsyncSessionDep,
):
"""
Transfère un élève d'une classe à une autre.
"""
# Récupérer l'élève
student_query = select(Student).where(Student.id == transfer_data.student_id)
student_result = await session.execute(student_query)
student = student_result.scalar_one_or_none()
if not student:
raise HTTPException(status_code=404, detail="Élève non trouvé")
# Vérifier que la nouvelle classe existe
new_class_query = select(ClassGroup).where(ClassGroup.id == transfer_data.new_class_group_id)
new_class_result = await session.execute(new_class_query)
new_class = new_class_result.scalar_one_or_none()
if not new_class:
raise HTTPException(status_code=404, detail="Nouvelle classe non trouvée")
# Trouver l'inscription active
active_query = (
select(StudentEnrollment)
.options(selectinload(StudentEnrollment.class_group))
.where(
StudentEnrollment.student_id == transfer_data.student_id,
StudentEnrollment.departure_date.is_(None)
)
)
active_result = await session.execute(active_query)
old_enrollment = active_result.scalar_one_or_none()
if not old_enrollment:
raise HTTPException(
status_code=400,
detail=f"Aucune inscription active trouvée pour l'élève {student.first_name} {student.last_name}"
)
old_class_name = old_enrollment.class_group.name
# Terminer l'ancienne inscription
old_enrollment.departure_date = transfer_data.transfer_date
old_enrollment.departure_reason = transfer_data.transfer_reason or f"Transfert vers {new_class.name}"
# Créer la nouvelle inscription
new_enrollment = StudentEnrollment(
student_id=transfer_data.student_id,
class_group_id=transfer_data.new_class_group_id,
enrollment_date=transfer_data.transfer_date,
enrollment_reason=transfer_data.transfer_reason or f"Transfert depuis {old_class_name}"
)
session.add(new_enrollment)
await session.commit()
await session.refresh(new_enrollment)
return TransferResponse(
old_enrollment_id=old_enrollment.id,
new_enrollment_id=new_enrollment.id,
student_name=f"{student.first_name} {student.last_name}",
old_class_name=old_class_name,
new_class_name=new_class.name,
message=f"Élève {student.first_name} {student.last_name} transféré de {old_class_name} vers {new_class.name}"
)
@router.post("/departure", response_model=DepartureResponse)
async def record_departure(
departure_data: DepartureRequest,
session: AsyncSessionDep,
):
"""
Enregistre le départ d'un élève de sa classe actuelle.
"""
# Récupérer l'élève
student_query = select(Student).where(Student.id == departure_data.student_id)
student_result = await session.execute(student_query)
student = student_result.scalar_one_or_none()
if not student:
raise HTTPException(status_code=404, detail="Élève non trouvé")
# Trouver l'inscription active
active_query = (
select(StudentEnrollment)
.options(selectinload(StudentEnrollment.class_group))
.where(
StudentEnrollment.student_id == departure_data.student_id,
StudentEnrollment.departure_date.is_(None)
)
)
active_result = await session.execute(active_query)
enrollment = active_result.scalar_one_or_none()
if not enrollment:
raise HTTPException(
status_code=400,
detail=f"Aucune inscription active trouvée pour l'élève {student.first_name} {student.last_name}"
)
class_name = enrollment.class_group.name
# Enregistrer le départ
enrollment.departure_date = departure_data.departure_date
enrollment.departure_reason = departure_data.departure_reason
await session.commit()
return DepartureResponse(
enrollment_id=enrollment.id,
student_name=f"{student.first_name} {student.last_name}",
class_name=class_name,
message=f"Départ de {student.first_name} {student.last_name} de {class_name} enregistré"
)