Files
MABEA/backend/scripts/eskalation_pruefen.py
patrickandClaude Sonnet 5 1afac3a797
CI / backend-tests (push) Failing after 1m56s
CI / frontend-build (push) Successful in 17s
fix(models): app/models/__init__.py fehlten 6 Module seit Personal-Modul
Nicht mitgezogen seit Migration 0012: personal, mangel, fahrzeugdetails,
lagerbewegung, dokument, permission. Live-Testlauf von scripts/create_admin.py
brach mit NoReferencedTableError ab (Benutzer.einheit_id -> einheit war nicht
registriert, da das Skript nur app.models.auth statt des vollständigen
Registry-Imports nutzt). App selbst lief nur zufällig fehlerfrei, weil
api.py alle Endpunkt-Module importiert und die dortigen Modell-Importe die
Lücke im Betrieb kaschiert haben.

Fix: __init__.py vervollständigt + create_admin.py/eskalation_pruefen.py
(täglicher Timer!) bekommen zusätzlich ein explizites "import app.models" als
Schutz, unabhängig davon ob die Registry künftig wieder unvollständig wird.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KC8HYvv6UkCVYheYiTw9DD
2026-09-05 14:09:30 +02:00

39 lines
1.1 KiB
Python

"""Karte 12: periodischer Eskalations-Prüflauf, für einen systemd-Timer gedacht.
Deployment-Regel (Prompt 19): nur Skript-Datei hier, Ausführung auf Zielsystem.
Direkter DB-Zugriff statt HTTP-Aufruf des Endpunkts (POST /eskalation/pruefen) -
kein Admin-Token/Login nötig für einen lokalen Cron/Timer auf demselben Host,
gleiches Muster wie scripts/create_admin.py.
Aufruf (im aktivierten venv auf dem Server):
python -m scripts.eskalation_pruefen
"""
from __future__ import annotations
import asyncio
import logging
import app.models # noqa: F401 (alle ORM-Modelle registrieren, siehe create_admin.py)
from app.db.session import SessionLocal
from app.services.eskalation import pruefe_offene_fehlbestaende
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def main() -> None:
async with SessionLocal() as db:
ergebnis = await pruefe_offene_fehlbestaende(db)
await db.commit()
logger.info(
"Eskalations-Prüflauf: %d Erinnerung(en), %d Eskalation(en) verschickt.",
ergebnis["erinnerungen"],
ergebnis["eskalationen"],
)
if __name__ == "__main__":
asyncio.run(main())