from __future__ import annotations from typing import Any import json import logging import shutil import uuid from datetime import UTC, datetime from pathlib import Path from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, Response, UploadFile from fastapi.encoders import jsonable_encoder from fastapi.responses import FileResponse, HTMLResponse, JSONResponse from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError, ProgrammingError from sqlalchemy.orm import Session from app.api.dependencies import current_admin, current_user from app.db.session import get_session from app.models.contact import Contact from app.models.customer import Customer from app.models.device import Device from app.models.equipment import Equipment from app.models.location import Location from app.models.user import User, UserRole from app.models.validation import Validation from app.modules.helios.service import HeliosImportService from app.modules.orion.service import OrionReportService from app.schemas.common import PaginatedResponse from app.schemas.domain import ( ContactCreate, ContactRead, ContactUpdate, CustomerCreate, CustomerRead, CustomerUpdate, DeviceCreate, DeviceRead, DeviceUpdate, EquipmentCreate, EquipmentRead, EquipmentUpdate, LocationCreate, LocationRead, LocationUpdate, MeasurementImportConfirmRequest, MeasurementImportPreviewRead, ValidationCreate, ValidationImportPreview, ValidationImportRequest, ValidationImportSummary, ValidationRead, ValidationReview, ValidationUpdate, QuickStartCreateRequest, QuickStartCustomerData, QuickStartValidationSummary, UserCreate, UserRead, UserUpdate, UserPasswordResetRequest, ) from app.services.domain_service import CrudService, DomainServices from app.services.quickstart_service import QuickStartService from app.services.validation_workflow import ValidationWorkflowService router = APIRouter(tags=["domain"], dependencies=[Depends(current_user)]) logger = logging.getLogger(__name__) @router.get("/report-templates/default/checklists") def default_report_checklists(session: Session = Depends(get_session)) -> list[dict]: from app.modules.orion.template_service import ReportTemplateService bundle = ReportTemplateService(session).ensure_default_template() session.commit() return [ { "checklist_key": item.checklist_key, "title": item.title, "columns": item.columns, "items": item.items, "order_index": item.order_index, } for item in bundle.checklists ] @router.get("/dashboard") def dashboard(session: Session = Depends(get_session)) -> dict[str, int]: today = func.current_date() result_groups = { "validation_result_open": ["OFFEN", "offen", None, ""], "validation_result_passed": ["BESTANDEN", "bestanden"], "validation_result_conditional": ["BESTANDEN_MIT_AUFLAGEN", "bestanden_mit_auflagen", "mit_auflagen", "bestanden mit Auflagen"], "validation_result_failed": ["NICHT_BESTANDEN", "nicht_bestanden", "nicht bestanden"], } result_counts = { key: session.scalar(select(func.count()).select_from(Validation).where(Validation.result.in_(values))) or 0 for key, values in result_groups.items() if None not in values } result_counts["validation_result_open"] = ( session.scalar( select(func.count()) .select_from(Validation) .where((Validation.result.in_(["OFFEN", "offen", ""])) | (Validation.result.is_(None))) ) or 0 ) return { "customers": session.scalar(select(func.count()).select_from(Customer)) or 0, "locations": session.scalar(select(func.count()).select_from(Location)) or 0, "contacts": session.scalar(select(func.count()).select_from(Contact)) or 0, "devices": session.scalar(select(func.count()).select_from(Device)) or 0, "equipment": session.scalar(select(func.count()).select_from(Equipment)) or 0, "equipment_green": session.scalar( select(func.count()).select_from(Equipment).where(Equipment.status == "green") ) or 0, "equipment_yellow": session.scalar( select(func.count()).select_from(Equipment).where(Equipment.status == "yellow") ) or 0, "equipment_red": session.scalar( select(func.count()).select_from(Equipment).where(Equipment.status == "red") ) or 0, "validations": session.scalar(select(func.count()).select_from(Validation)) or 0, "validation_drafts": session.scalar( select(func.count()).select_from(Validation).where(Validation.status == "ENTWURF") ) or 0, "validation_ready": session.scalar( select(func.count()) .select_from(Validation) .where(Validation.status == "BEREIT_ZUR_PRUEFUNG") ) or 0, "validation_in_review": session.scalar( select(func.count()).select_from(Validation).where(Validation.status == "IN_PRUEFUNG") ) or 0, "validation_approved": session.scalar( select(func.count()).select_from(Validation).where(Validation.status == "FREIGEGEBEN") ) or 0, "validation_completed": session.scalar( select(func.count()).select_from(Validation).where(Validation.status == "ABGESCHLOSSEN") ) or 0, "validation_overdue": session.scalar( select(func.count()) .select_from(Validation) .where(Validation.next_validation_on < today) ) or 0, **result_counts, } def paging( page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), search: str | None = Query(default=None, max_length=120), ) -> dict[str, Any]: return {"page": page, "page_size": page_size, "search": search} def commit_create(session: Session, service: CrudService, payload): try: item = service.create(payload.model_dump()) if isinstance(item, Validation): ValidationWorkflowService(session).apply_revalidation_date(item) session.commit() session.refresh(item) return item except IntegrityError as exc: session.rollback() from fastapi import HTTPException raise HTTPException( status_code=409, detail="Datensatz verletzt Datenbankbeziehungen." ) from exc def commit_update(session: Session, service: CrudService, item_id: str, payload): try: item = service.update(item_id, payload.model_dump()) if isinstance(item, Validation): ValidationWorkflowService(session).apply_revalidation_date(item) session.commit() session.refresh(item) return item except IntegrityError as exc: session.rollback() from fastapi import HTTPException raise HTTPException( status_code=409, detail="Datensatz verletzt Datenbankbeziehungen." ) from exc def commit_delete(session: Session, service: CrudService, item_id: str) -> Response: service.delete(item_id) session.commit() return Response(status_code=204) @router.get("/customers", response_model=PaginatedResponse[CustomerRead]) def list_customers(params: dict = Depends(paging), session: Session = Depends(get_session)): return DomainServices(session).customers.list(**params) @router.post("/customers", response_model=CustomerRead, status_code=201) def create_customer(payload: CustomerCreate, session: Session = Depends(get_session)): return commit_create(session, DomainServices(session).customers, payload) @router.put("/customers/{item_id}", response_model=CustomerRead) def update_customer(item_id: str, payload: CustomerUpdate, session: Session = Depends(get_session)): return commit_update(session, DomainServices(session).customers, item_id, payload) @router.delete("/customers/{item_id}", status_code=204) def delete_customer(item_id: str, session: Session = Depends(get_session)): return commit_delete(session, DomainServices(session).customers, item_id) @router.get("/customers/{item_id}/quickstart-data", response_model=QuickStartCustomerData) def customer_quickstart_data(item_id: str, session: Session = Depends(get_session)): try: return QuickStartService(session).customer_data(item_id) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @router.get("/locations", response_model=PaginatedResponse[LocationRead]) def list_locations(params: dict = Depends(paging), session: Session = Depends(get_session)): return DomainServices(session).locations.list(**params) @router.post("/locations", response_model=LocationRead, status_code=201) def create_location(payload: LocationCreate, session: Session = Depends(get_session)): return commit_create(session, DomainServices(session).locations, payload) @router.put("/locations/{item_id}", response_model=LocationRead) def update_location(item_id: str, payload: LocationUpdate, session: Session = Depends(get_session)): return commit_update(session, DomainServices(session).locations, item_id, payload) @router.delete("/locations/{item_id}", status_code=204) def delete_location(item_id: str, session: Session = Depends(get_session)): return commit_delete(session, DomainServices(session).locations, item_id) @router.get("/contacts", response_model=PaginatedResponse[ContactRead]) def list_contacts(params: dict = Depends(paging), session: Session = Depends(get_session)): return DomainServices(session).contacts.list(**params) @router.post("/contacts", response_model=ContactRead, status_code=201) def create_contact(payload: ContactCreate, session: Session = Depends(get_session)): return commit_create(session, DomainServices(session).contacts, payload) @router.put("/contacts/{item_id}", response_model=ContactRead) def update_contact(item_id: str, payload: ContactUpdate, session: Session = Depends(get_session)): return commit_update(session, DomainServices(session).contacts, item_id, payload) @router.delete("/contacts/{item_id}", status_code=204) def delete_contact(item_id: str, session: Session = Depends(get_session)): return commit_delete(session, DomainServices(session).contacts, item_id) @router.get("/devices", response_model=PaginatedResponse[DeviceRead]) def list_devices(params: dict = Depends(paging), session: Session = Depends(get_session)): return DomainServices(session).devices.list(**params) @router.post("/devices", response_model=DeviceRead, status_code=201) def create_device(payload: DeviceCreate, session: Session = Depends(get_session)): return commit_create(session, DomainServices(session).devices, payload) @router.put("/devices/{item_id}", response_model=DeviceRead) def update_device(item_id: str, payload: DeviceUpdate, session: Session = Depends(get_session)): return commit_update(session, DomainServices(session).devices, item_id, payload) @router.delete("/devices/{item_id}", status_code=204) def delete_device(item_id: str, session: Session = Depends(get_session)): return commit_delete(session, DomainServices(session).devices, item_id) @router.get("/equipment", response_model=PaginatedResponse[EquipmentRead]) def list_equipment(params: dict = Depends(paging), session: Session = Depends(get_session)): return DomainServices(session).equipment.list(**params) @router.post("/equipment", response_model=EquipmentRead, status_code=201) def create_equipment(payload: EquipmentCreate, session: Session = Depends(get_session)): return commit_create(session, DomainServices(session).equipment, payload) @router.put("/equipment/{item_id}", response_model=EquipmentRead) def update_equipment( item_id: str, payload: EquipmentUpdate, session: Session = Depends(get_session) ): return commit_update(session, DomainServices(session).equipment, item_id, payload) @router.delete("/equipment/{item_id}", status_code=204) def delete_equipment(item_id: str, session: Session = Depends(get_session)): return commit_delete(session, DomainServices(session).equipment, item_id) @router.get("/validations", response_model=PaginatedResponse[ValidationRead]) def list_validations( search: str | None = Query(default=None, max_length=120), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), sort_by: str = Query(default="updated_at"), sort_order: str = Query(default="desc", pattern="^(asc|desc)$"), status: str | None = None, customer_id: str | None = None, device_id: str | None = None, validation_type: str | None = None, result: str | None = None, date_from: str | None = None, date_to: str | None = None, overdue_only: bool = False, session: Session = Depends(get_session), ): return ValidationWorkflowService(session).query_validations( search=search, page=page, page_size=page_size, sort_by=sort_by, sort_order=sort_order, filters={ "status": status, "customer_id": customer_id, "device_id": device_id, "validation_type": validation_type, "result": result, "date_from": date_from, "date_to": date_to, "overdue_only": overdue_only, }, ) @router.get("/validations/next-report-number") def next_report_number(session: Session = Depends(get_session)) -> dict[str, str]: count = session.scalar(select(func.count()).select_from(Validation)) or 0 return {"report_number": f"VAL-{count + 1:05d}"} @router.get("/validations/{item_id}", response_model=ValidationRead) def get_validation(item_id: str, session: Session = Depends(get_session)): item = DomainServices(session).validations.repository.get(item_id) if item is None: from fastapi import HTTPException raise HTTPException(status_code=404, detail="Resource not found") return item @router.get("/devices/{item_id}/last-validation", response_model=QuickStartValidationSummary | None) def last_device_validation(item_id: str, session: Session = Depends(get_session)): try: return QuickStartService(session).device_last_validation(item_id) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @router.post("/validations", response_model=ValidationRead, status_code=201) def create_validation(payload: ValidationCreate, session: Session = Depends(get_session)): return commit_create(session, DomainServices(session).validations, payload) @router.post("/validations/quick-start", response_model=ValidationRead, status_code=201) def create_quick_start_validation( payload: QuickStartCreateRequest, session: Session = Depends(get_session), me: User = Depends(current_user), ): try: validation = QuickStartService(session).create_validation(payload, me) session.commit() session.refresh(validation) return validation except ValueError as exc: session.rollback() raise HTTPException(status_code=422, detail=str(exc)) from exc @router.put("/validations/{item_id}", response_model=ValidationRead) def update_validation( item_id: str, payload: ValidationUpdate, session: Session = Depends(get_session) ): return commit_update(session, DomainServices(session).validations, item_id, payload) @router.delete("/validations/{item_id}", status_code=204) def delete_validation(item_id: str, session: Session = Depends(get_session)): item = DomainServices(session).validations.repository.get(item_id) if item and item.status != "ENTWURF": from fastapi import HTTPException raise HTTPException(status_code=409, detail="Only draft validations can be deleted") return commit_delete(session, DomainServices(session).validations, item_id) @router.post("/validations/{item_id}/review", response_model=ValidationReview) def review_validation(item_id: str, session: Session = Depends(get_session)): item = DomainServices(session).validations.repository.get(item_id) if item is None: from fastapi import HTTPException raise HTTPException(status_code=404, detail="Resource not found") review = ValidationWorkflowService(session).mark_ready_for_review(item) session.commit() return review @router.post("/validations/{item_id}/duplicate", response_model=ValidationRead, status_code=201) def duplicate_validation(item_id: str, session: Session = Depends(get_session)): item = DomainServices(session).validations.repository.get(item_id) if item is None: from fastapi import HTTPException raise HTTPException(status_code=404, detail="Resource not found") clone = ValidationWorkflowService(session).duplicate(item) session.commit() session.refresh(clone) return clone @router.post("/validations/{item_id}/new-version", response_model=ValidationRead, status_code=201) def new_validation_version(item_id: str, session: Session = Depends(get_session)): item = DomainServices(session).validations.repository.get(item_id) if item is None: from fastapi import HTTPException raise HTTPException(status_code=404, detail="Resource not found") clone = ValidationWorkflowService(session).create_new_version(item) session.commit() session.refresh(clone) return clone @router.post("/validations/{item_id}/cancel", response_model=ValidationRead) def cancel_validation(item_id: str, session: Session = Depends(get_session)): item = DomainServices(session).validations.repository.get(item_id) if item is None: from fastapi import HTTPException raise HTTPException(status_code=404, detail="Resource not found") item.status = "STORNIERT" session.commit() session.refresh(item) return item @router.post("/validations/{item_id}/attachments") def upload_validation_attachment( item_id: str, category: str = Form(...), description: str = Form(default=""), order: int = Form(default=0), file: UploadFile = File(...), session: Session = Depends(get_session), ): item = DomainServices(session).validations.repository.get(item_id) if item is None: from fastapi import HTTPException raise HTTPException(status_code=404, detail="Resource not found") if item.status in {"FREIGEGEBEN", "ABGESCHLOSSEN"}: from fastapi import HTTPException raise HTTPException(status_code=409, detail="Freigegebene Berichte sind schreibgeschuetzt.") original_name = Path(file.filename or "anlage").name suffix = Path(original_name).suffix stored_name = f"{uuid.uuid4().hex}{suffix}" upload_dir = Path("/app/uploads/validations") / item_id upload_dir.mkdir(parents=True, exist_ok=True) storage_path = upload_dir / stored_name with storage_path.open("wb") as target: shutil.copyfileobj(file.file, target) attachment = { "category": category, "filename": original_name, "content_type": file.content_type, "description": description, "order": order, "storage_path": str(storage_path), "url": f"/uploads/validations/{item_id}/{stored_name}", } current = list(item.attachments or []) current.append(attachment) item.attachments = current session.commit() return attachment @router.post( "/validations/{item_id}/measurement-imports/winlog-pdf", response_model=MeasurementImportPreviewRead, ) async def upload_winlog_pdf_import( item_id: str, file: UploadFile = File(...), attachment_only: bool = Form(default=False), session: Session = Depends(get_session), ): item = DomainServices(session).validations.repository.get(item_id) if item is None: from fastapi import HTTPException raise HTTPException(status_code=404, detail="Resource not found") preview = HeliosImportService(session).save_winlog_pdf( item_id, file.filename or "winlog.pdf", await file.read(), attachment_only=attachment_only, ) session.commit() return preview @router.post( "/measurement-imports/{import_id}/confirm", response_model=MeasurementImportPreviewRead, ) def confirm_measurement_import( import_id: str, payload: MeasurementImportConfirmRequest, session: Session = Depends(get_session), ): try: preview = HeliosImportService(session).confirm_values( import_id, [item.model_dump() for item in payload.values] ) except ValueError as exc: from fastapi import HTTPException raise HTTPException(status_code=404, detail=str(exc)) from exc session.commit() return preview @router.get("/validations/{item_id}/export.json") def export_validation_json(item_id: str, session: Session = Depends(get_session)): item = DomainServices(session).validations.repository.get(item_id) if item is None: from fastapi import HTTPException raise HTTPException(status_code=404, detail="Resource not found") data = ValidationWorkflowService(session).export_json(item) return JSONResponse( content=json.loads(json.dumps(data, default=str)), headers={"Content-Disposition": f'attachment; filename="{item.report_number}.json"'}, ) @router.post("/validations/import/csv-preview", response_model=ValidationImportPreview) async def preview_validation_csv(file: UploadFile, session: Session = Depends(get_session)): return ValidationWorkflowService(session).preview_csv(await file.read()) @router.post("/validations/import/json", response_model=ValidationImportSummary) def import_validation_json( payload: ValidationImportRequest, session: Session = Depends(get_session) ): summary = ValidationWorkflowService(session).import_rows( payload.rows, payload.duplicate_strategy ) session.commit() return summary @router.get("/validations/{item_id}/report.html", response_class=HTMLResponse) def validation_report_preview(item_id: str, session: Session = Depends(get_session)): return OrionReportService(session, Path("/app/reports")).render_html(item_id) @router.get("/validations/{item_id}/report.pdf") def validation_report_pdf(item_id: str, session: Session = Depends(get_session)): try: report_path = OrionReportService(session, Path("/app/reports")).render_pdf(item_id) except Exception as exc: logger.exception("PDF generation failed for validation %s", item_id) from fastapi import HTTPException raise HTTPException( status_code=502, detail="Der PDF-Bericht konnte nicht erzeugt werden. Die HTML-Vorschau ist weiterhin verfuegbar.", ) from exc return FileResponse( report_path, media_type="application/pdf", filename=report_path.name, ) def _user_payload(user: User) -> dict: return { "id": user.id, "created_at": user.created_at, "updated_at": user.updated_at, "email": user.email, "first_name": user.first_name, "last_name": user.last_name, "role": UserRole(user.role), "is_active": user.is_active, "must_change_password": user.must_change_password, "last_login_at": user.last_login_at, "password_changed_at": user.password_changed_at, } @router.get("/users", response_model=PaginatedResponse[UserRead]) def list_users( session: Session = Depends(get_session), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), search: str | None = Query(default=None), role: str | None = Query(default=None), active: bool | None = Query(default=None), sort_by: str = Query(default="created_at"), sort_order: str = Query(default="desc"), _: User = Depends(current_admin), ): query = select(User) query = query.where(User.deleted_at.is_(None)) if search: term = f"%{search.strip()}%" query = query.where( (User.first_name.ilike(term)) | (User.last_name.ilike(term)) | (User.email.ilike(term)) ) if role: query = query.where(User.role == role) if active is not None: query = query.where(User.is_active.is_(active)) sort_columns = { "name": User.last_name, "email": User.email, "role": User.role, "is_active": User.is_active, "last_login_at": User.last_login_at, "password_changed_at": User.password_changed_at, "created_at": User.created_at, "updated_at": User.updated_at, } sort_column = sort_columns.get(sort_by, User.created_at) query = query.order_by(sort_column.asc() if sort_order.lower() == "asc" else sort_column.desc()) total = session.scalar(select(func.count()).select_from(query.order_by(None).subquery())) or 0 pages = max((total + page_size - 1) // page_size, 1) items = list(session.scalars(query.offset((page - 1) * page_size).limit(page_size))) return {"items": items, "total": total, "page": page, "page_size": page_size, "pages": pages} @router.post("/users", status_code=201) def create_user(payload: UserCreate, session: Session = Depends(get_session), _: User = Depends(current_admin)): from app.core.security import hash_password import secrets from app.models.user import User normalized_email = payload.email.strip().lower() logger.debug("create_user request email=%s normalized_email=%s", payload.email, normalized_email) existing = session.scalar(select(User).where(func.lower(User.email) == normalized_email)) logger.debug("create_user select result=%s count=%s", bool(existing), 1 if existing else 0) if existing is not None: raise HTTPException(status_code=409, detail="Die E-Mail-Adresse wird bereits verwendet.") temporary_password = payload.temporary_password or secrets.token_urlsafe(12) try: user = User( email=normalized_email, first_name=payload.first_name, last_name=payload.last_name, role=payload.role.value, password_hash=hash_password(temporary_password), is_active=payload.is_active, must_change_password=payload.must_change_password, ) logger.debug("create_user insert email=%s", normalized_email) session.add(user) session.commit() logger.debug("create_user commit ok email=%s id=%s", normalized_email, user.id) session.refresh(user) logger.debug("create_user response email=%s status=201", normalized_email) return JSONResponse( content={"temporary_password": temporary_password, "user": jsonable_encoder(user)}, status_code=201, ) except IntegrityError as exc: session.rollback() logger.exception("create_user integrity error email=%s", normalized_email) raise HTTPException(status_code=409, detail="Die E-Mail-Adresse wird bereits verwendet.") from exc except ProgrammingError as exc: session.rollback() raise HTTPException(status_code=422, detail="Die Benutzerrolle ist ungültig.") from exc @router.get("/users/{item_id}", response_model=UserRead) def get_user(item_id: str, session: Session = Depends(get_session), _: User = Depends(current_admin)): user = session.get(User, item_id) if user is None or user.deleted_at is not None: raise HTTPException(status_code=404, detail="Resource not found") return user @router.put("/users/{item_id}", response_model=UserRead) def update_user(item_id: str, payload: UserUpdate, session: Session = Depends(get_session), me: User = Depends(current_admin)): user = session.get(User, item_id) if user is None or user.deleted_at is not None: raise HTTPException(status_code=404, detail="Resource not found") if user.id == me.id and not payload.is_active: raise HTTPException(status_code=409, detail="Own admin account cannot be deactivated") active_admins = session.scalar( select(func.count()).select_from(User).where(User.role == UserRole.ADMIN.value, User.is_active.is_(True)) ) or 0 if user.role == UserRole.ADMIN.value and active_admins <= 1 and (not payload.is_active or payload.role != UserRole.ADMIN.value): raise HTTPException(status_code=409, detail="Der letzte aktive ADMIN darf nicht deaktiviert oder herabgestuft werden.") try: normalized_email = payload.email.strip().lower() logger.debug("update_user request id=%s email=%s normalized_email=%s", item_id, payload.email, normalized_email) existing = session.scalar( select(User).where(func.lower(User.email) == normalized_email, User.id != user.id) ) logger.debug("update_user select result=%s count=%s", bool(existing), 1 if existing else 0) if existing is not None: raise HTTPException(status_code=409, detail="Die E-Mail-Adresse wird bereits verwendet.") user.first_name = payload.first_name user.last_name = payload.last_name user.email = normalized_email user.role = payload.role.value user.is_active = payload.is_active user.must_change_password = payload.must_change_password logger.debug("update_user commit start id=%s email=%s", user.id, normalized_email) session.commit() logger.debug("update_user commit ok id=%s email=%s", user.id, normalized_email) session.refresh(user) return user except IntegrityError as exc: session.rollback() logger.exception("update_user integrity error id=%s email=%s", item_id, payload.email) raise HTTPException(status_code=409, detail="Die E-Mail-Adresse wird bereits verwendet.") from exc except ProgrammingError as exc: session.rollback() raise HTTPException(status_code=422, detail="Die Benutzerrolle ist ungültig.") from exc @router.delete("/users/{item_id}", status_code=204) def delete_user(item_id: str, session: Session = Depends(get_session), me: User = Depends(current_admin)): user = session.get(User, item_id) if user is None or user.deleted_at is not None: raise HTTPException(status_code=404, detail="Resource not found") if user.id == me.id: raise HTTPException(status_code=409, detail="Das eigene Benutzerkonto kann nicht geloescht werden.") active_admins = session.scalar( select(func.count()).select_from(User).where(User.role == UserRole.ADMIN.value, User.is_active.is_(True)) ) or 0 if user.role == UserRole.ADMIN.value and active_admins <= 1: raise HTTPException(status_code=409, detail="Der letzte aktive ADMIN darf nicht geloescht werden.") referenced_validations = session.scalar( select(func.count()).select_from(Validation).where(Validation.examiner_id == user.id) ) or 0 if referenced_validations == 0: session.delete(user) session.commit() return Response(status_code=204) user.is_active = False user.deleted_at = datetime.now(UTC) session.commit() return Response(status_code=204) @router.post("/users/{item_id}/reset-password") def reset_user_password(item_id: str, session: Session = Depends(get_session), _: User = Depends(current_admin)): import secrets from app.core.security import hash_password user = session.get(User, item_id) if user is None or user.deleted_at is not None: raise HTTPException(status_code=404, detail="Resource not found") temporary_password = secrets.token_urlsafe(12) user.password_hash = hash_password(temporary_password) user.must_change_password = True session.commit() return JSONResponse(content={"temporary_password": temporary_password})