Backend: gemeinsame Helfer in common.py

- Eine ValidationError-Klasse statt zwei (links.py, projects.py), ein
  Fehler-Handler in main.py
- clean_text() ersetzt vier fast gleiche Text-Prüfungen (Kanban, Haushalt,
  Reisen, Artikel); Fehlermeldungen nennen jetzt überall die Höchstlänge
- Zeitstempel (now_iso, iso_ago, iso_after) und Kontenliste (user_list)
  an einer Stelle; ungenutzte Importe entfernt
- Tests für die Helfer und für einheitliche 400-Fehler aus allen Bereichen

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
guyg1474andClaude Opus 5.5 committed 2026-10-15 18:10:43 +02:00
1 parent d28e6b2e4b
commit 4aba48ce51
14 files changed
+117 -86

No files matched your search

+9 -29
View File
@@ -8,14 +8,13 @@ Erledigte Aufgaben wandern nach 2 Tagen ins Archiv und werden nach 14 Tagen gel
import asyncio
import logging
import math
import re
from datetime import datetime, timedelta, timezone
from datetime import timedelta
from fastapi import APIRouter, Body, Depends, HTTPException
from .auth import current_user
from .db import now_iso, transaction
from .links import ValidationError
from .common import ValidationError, clean_text, iso_after, iso_ago, now_iso, user_list
from .db import transaction
router = APIRouter()
@@ -33,19 +32,10 @@ CLEANUP_SECONDS = 3600
log = logging.getLogger(__name__)
def _ago(delta: timedelta) -> str:
# Gleiches Format wie now_iso(), damit Textvergleich in SQLite stimmt
return (datetime.now(timezone.utc) - delta).isoformat(timespec="milliseconds").replace("+00:00", "Z")
def _iso_after(iso: str, delta: timedelta) -> str:
return (datetime.fromisoformat(iso.replace("Z", "+00:00")) + delta).isoformat(timespec="milliseconds").replace("+00:00", "Z")
def cleanup(conn) -> int:
"""Seit 14 Tagen erledigte Aufgaben löschen."""
return conn.execute(
"DELETE FROM kanban_tasks WHERE status = 'erledigt' AND done_at <= ?", (_ago(DELETE_AFTER),)
"DELETE FROM kanban_tasks WHERE status = 'erledigt' AND done_at <= ?", (iso_ago(DELETE_AFTER),)
).rowcount
@@ -72,12 +62,7 @@ def _may_delete(user: dict, created_by) -> bool:
def _name(value, field="Name", max_len=MAX_NAME) -> str:
v = re.sub(r"\s+", " ", value).strip() if isinstance(value, str) else ""
if not v:
raise ValidationError(f"{field} fehlt.")
if len(v) > max_len:
raise ValidationError(f"{field} ist zu lang (max. {max_len} Zeichen).")
return v
return clean_text(value, field, max_len)
def _description(value) -> str:
@@ -146,11 +131,6 @@ def _next(conn, table: str, where: str = "", args: tuple = ()) -> float:
# ---------- Ausgabe ----------
def _users(conn) -> list:
rows = conn.execute("SELECT id, display_name FROM users ORDER BY display_name COLLATE NOCASE").fetchall()
return [{"id": u["id"], "name": u["display_name"]} for u in rows]
def _project_links(conn) -> dict:
links = {}
for r in conn.execute(
@@ -199,7 +179,7 @@ def _overview(conn) -> dict:
return {
"projects": _projects(conn),
"articles": [{"id": a["id"], "title": a["title"]} for a in articles],
"users": _users(conn),
"users": user_list(conn),
}
@@ -236,7 +216,7 @@ def _board(conn, project_id: int) -> dict:
(project_id,),
).fetchall()
articles = conn.execute("SELECT id, title FROM projects ORDER BY title COLLATE NOCASE").fetchall()
archive_before = _ago(ARCHIVE_AFTER)
archive_before = iso_ago(ARCHIVE_AFTER)
archived = sorted(
(r for r in tasks if r["status"] == "erledigt" and r["done_at"] and r["done_at"] <= archive_before),
key=lambda r: r["done_at"],
@@ -247,8 +227,8 @@ def _board(conn, project_id: int) -> dict:
"project": project,
"tasks": [_task(r) for r in tasks if r["id"] not in archived_ids],
# Archiv: nur mit eigenem Button sichtbar; deleteAt = automatisches Löschen
"archived": [{**_task(r), "deleteAt": _iso_after(r["done_at"], DELETE_AFTER)} for r in archived],
"users": _users(conn),
"archived": [{**_task(r), "deleteAt": iso_after(r["done_at"], DELETE_AFTER)} for r in archived],
"users": user_list(conn),
"articles": [{"id": a["id"], "title": a["title"]} for a in articles],
}