Karte 12: systemd-Timer für den täglichen Eskalations-Prüflauf
CI / backend-tests (push) Successful in 1m8s
CI / frontend-build (push) Successful in 25s

scripts/eskalation_pruefen.py ruft den Service direkt gegen die DB auf (kein
HTTP/Admin-Token nötig, gleiches Muster wie scripts/create_admin.py).
mabea-eskalation.timer läuft täglich 06:00 Uhr. Zeitschwellen bleiben in der
DB (eskalation_konfiguration), nicht im Timer.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
This commit is contained in:
2026-09-04 18:28:46 +02:00
co-authored by Claude Sonnet 5
parent e0cbf4b8de
commit c71df1ac4d
5 changed files with 145 additions and 0 deletions
+37
View File
@@ -0,0 +1,37 @@
"""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
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())