guyg1474andClaude Opus 5.5
dd58b3d1f8
FastAPI-Backend und React-Frontend mit Konten, Sendungsverfolgung, Projekt-Artikeln, Vorrat & Einkauf mit Rezeptabgleich, Reise-Weltkarte, Kanban-Projekten mit Priorität, Server-Monitoring und den Designs Matrix und Blackwall. Betrieb per Docker hinter nginx (siehe DEPLOY.md). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
369 lines
15 KiB
Python
369 lines
15 KiB
Python
"""Kanban: Projekte (je ein Board) → Aufgaben mit Status, Bearbeiter und Beschreibung.
|
||
|
||
Alles ist für alle Konten gemeinsam. Löschen dürfen nur, wer etwas angelegt hat, und Administratoren.
|
||
Projekte lassen sich als abgeschlossen markieren und mit Artikeln aus "P42 Projekte" verknüpfen.
|
||
`position` (Kommazahl) sortiert Projekte und Aufgaben innerhalb einer Spalte.
|
||
"""
|
||
import math
|
||
import re
|
||
|
||
from fastapi import APIRouter, Body, Depends, HTTPException
|
||
|
||
from .auth import current_user
|
||
from .db import now_iso, transaction
|
||
from .links import ValidationError
|
||
|
||
router = APIRouter()
|
||
|
||
STATUSES = ("offen", "arbeit", "wartet", "erledigt")
|
||
MAX_NAME = 80
|
||
MAX_TITLE = 120
|
||
MAX_DESCRIPTION = 5000
|
||
MAX_TASKS = 1000 # pro Projekt
|
||
MAX_PROJECTS = 300
|
||
MAX_LINKS = 20
|
||
|
||
|
||
def _person(user_id, name) -> dict | None:
|
||
return {"id": user_id, "name": name} if user_id is not None and name else None
|
||
|
||
|
||
def _may_delete(user: dict, created_by) -> bool:
|
||
return user["isAdmin"] or created_by == user["id"]
|
||
|
||
|
||
# ---------- Eingaben prüfen ----------
|
||
|
||
|
||
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
|
||
|
||
|
||
def _description(value) -> str:
|
||
if value is None:
|
||
return ""
|
||
if not isinstance(value, str):
|
||
raise ValidationError("Ungültige Beschreibung.")
|
||
text = value.replace("\r\n", "\n").strip()
|
||
if len(text) > MAX_DESCRIPTION:
|
||
raise ValidationError(f"Beschreibung ist zu lang (max. {MAX_DESCRIPTION} Zeichen).")
|
||
return text
|
||
|
||
|
||
PRIORITIES = {"hoch": 1, "mittel": 2, "niedrig": 3}
|
||
PRIORITY_NAMES = {v: k for k, v in PRIORITIES.items()}
|
||
|
||
|
||
def _priority(value) -> int:
|
||
if value not in PRIORITIES:
|
||
raise ValidationError("Unbekannte Priorität.")
|
||
return PRIORITIES[value]
|
||
|
||
|
||
def _status(value) -> str:
|
||
if value not in STATUSES:
|
||
raise ValidationError("Unbekannter Status.")
|
||
return value
|
||
|
||
|
||
def _assignee(conn, value):
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, bool) or not isinstance(value, int) or not conn.execute(
|
||
"SELECT 1 FROM users WHERE id = ?", (value,)
|
||
).fetchone():
|
||
raise ValidationError("Unbekannter Bearbeiter.")
|
||
return value
|
||
|
||
|
||
def _position(value) -> float:
|
||
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
|
||
raise ValidationError("Ungültige Position.")
|
||
return float(value)
|
||
|
||
|
||
def _links(conn, value) -> list:
|
||
if value is None:
|
||
return []
|
||
if not isinstance(value, list) or len(value) > MAX_LINKS or not all(isinstance(v, str) for v in value):
|
||
raise ValidationError("Ungültige Artikel-Verknüpfung.")
|
||
ids = list(dict.fromkeys(value))
|
||
found = {r["id"] for r in conn.execute(
|
||
f"SELECT id FROM projects WHERE id IN ({','.join('?' * len(ids))})", ids
|
||
)} if ids else set()
|
||
if len(found) != len(ids):
|
||
raise ValidationError("Ein verknüpfter Artikel existiert nicht mehr.")
|
||
return ids
|
||
|
||
|
||
def _next(conn, table: str, where: str = "", args: tuple = ()) -> float:
|
||
# table/where kommen nur aus diesem Modul, nie aus Anfragen
|
||
row = conn.execute(f"SELECT MAX(position) FROM {table} {where}", args).fetchone()
|
||
return (row[0] or 0) + 1
|
||
|
||
|
||
# ---------- 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(
|
||
"""SELECT l.project_id, p.id, p.title FROM kanban_project_links l
|
||
JOIN projects p ON p.id = l.article_id ORDER BY p.title COLLATE NOCASE"""
|
||
):
|
||
links.setdefault(r["project_id"], []).append({"id": r["id"], "title": r["title"]})
|
||
return links
|
||
|
||
|
||
def _projects(conn, project_id=None) -> list:
|
||
rows = conn.execute(
|
||
"""SELECT k.*, cu.display_name AS created_by_name, uu.display_name AS updated_by_name
|
||
FROM kanban_projects k
|
||
LEFT JOIN users cu ON cu.id = k.created_by
|
||
LEFT JOIN users uu ON uu.id = k.updated_by"""
|
||
+ (" WHERE k.id = ?" if project_id is not None else "")
|
||
+ " ORDER BY k.position",
|
||
(project_id,) if project_id is not None else (),
|
||
).fetchall()
|
||
counts = {}
|
||
for r in conn.execute("SELECT project_id, status, COUNT(*) AS n FROM kanban_tasks GROUP BY project_id, status"):
|
||
counts.setdefault(r["project_id"], dict.fromkeys(STATUSES, 0))[r["status"]] = r["n"]
|
||
links = _project_links(conn)
|
||
return [
|
||
{
|
||
"id": r["id"],
|
||
"name": r["name"],
|
||
"description": r["description"],
|
||
"position": r["position"],
|
||
"completedAt": r["completed_at"],
|
||
"links": links.get(r["id"], []),
|
||
"counts": counts.get(r["id"], dict.fromkeys(STATUSES, 0)),
|
||
"createdAt": r["created_at"],
|
||
"createdBy": _person(r["created_by"], r["created_by_name"]),
|
||
"updatedAt": r["updated_at"],
|
||
"updatedBy": _person(r["updated_by"], r["updated_by_name"]),
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
def _overview(conn) -> dict:
|
||
articles = conn.execute("SELECT id, title FROM projects ORDER BY title COLLATE NOCASE").fetchall()
|
||
return {
|
||
"projects": _projects(conn),
|
||
"articles": [{"id": a["id"], "title": a["title"]} for a in articles],
|
||
"users": _users(conn),
|
||
}
|
||
|
||
|
||
def _task(row) -> dict:
|
||
return {
|
||
"id": row["id"],
|
||
"projectId": row["project_id"],
|
||
"title": row["title"],
|
||
"description": row["description"],
|
||
"status": row["status"],
|
||
"assignee": row["assignee"],
|
||
"priority": PRIORITY_NAMES[row["priority"]],
|
||
"position": row["position"],
|
||
"createdAt": row["created_at"],
|
||
"createdBy": _person(row["created_by"], row["created_by_name"]),
|
||
"updatedAt": row["updated_at"],
|
||
"updatedBy": _person(row["updated_by"], row["updated_by_name"]),
|
||
}
|
||
|
||
|
||
def _board(conn, project_id: int) -> dict:
|
||
projects = _projects(conn, project_id)
|
||
if not projects:
|
||
raise HTTPException(404, "Kanban-Projekt nicht gefunden – vielleicht wurde es gerade gelöscht.")
|
||
project = projects[0]
|
||
tasks = conn.execute(
|
||
"""SELECT t.*, cu.display_name AS created_by_name, uu.display_name AS updated_by_name
|
||
FROM kanban_tasks t
|
||
LEFT JOIN users cu ON cu.id = t.created_by
|
||
LEFT JOIN users uu ON uu.id = t.updated_by
|
||
WHERE t.project_id = ? ORDER BY t.status, t.priority, t.position""",
|
||
(project_id,),
|
||
).fetchall()
|
||
articles = conn.execute("SELECT id, title FROM projects ORDER BY title COLLATE NOCASE").fetchall()
|
||
return {
|
||
"project": project,
|
||
"tasks": [_task(r) for r in tasks],
|
||
"users": _users(conn),
|
||
"articles": [{"id": a["id"], "title": a["title"]} for a in articles],
|
||
}
|
||
|
||
|
||
def _save_links(conn, project_id: int, links: list) -> None:
|
||
conn.execute("DELETE FROM kanban_project_links WHERE project_id = ?", (project_id,))
|
||
conn.executemany(
|
||
"INSERT INTO kanban_project_links (project_id, article_id) VALUES (?, ?)", [(project_id, a) for a in links]
|
||
)
|
||
|
||
|
||
# ---------- Übersicht ----------
|
||
|
||
|
||
@router.get("/api/kanban")
|
||
def get_overview(_: dict = Depends(current_user)):
|
||
with transaction() as conn:
|
||
return _overview(conn)
|
||
|
||
|
||
# ---------- Projekte ----------
|
||
|
||
|
||
@router.post("/api/kanban/projekte", status_code=201)
|
||
def create_project(data: dict = Body(...), user: dict = Depends(current_user)):
|
||
name = _name(data.get("name"))
|
||
description = _description(data.get("description"))
|
||
with transaction() as conn:
|
||
links = _links(conn, data.get("links"))
|
||
if conn.execute("SELECT COUNT(*) FROM kanban_projects").fetchone()[0] >= MAX_PROJECTS:
|
||
raise ValidationError(f"Maximal {MAX_PROJECTS} Kanban-Projekte.")
|
||
now = now_iso()
|
||
project_id = conn.execute(
|
||
"""INSERT INTO kanban_projects (name, description, position, created_at, created_by, updated_at, updated_by)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
||
(name, description, _next(conn, "kanban_projects"), now, user["id"], now, user["id"]),
|
||
).lastrowid
|
||
_save_links(conn, project_id, links)
|
||
return {"id": project_id, **_overview(conn)}
|
||
|
||
|
||
@router.get("/api/kanban/projekte/{project_id}")
|
||
def get_board(project_id: int, _: dict = Depends(current_user)):
|
||
with transaction() as conn:
|
||
return _board(conn, project_id)
|
||
|
||
|
||
@router.patch("/api/kanban/projekte/{project_id}")
|
||
def update_project(project_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
with transaction() as conn:
|
||
if not conn.execute("SELECT 1 FROM kanban_projects WHERE id = ?", (project_id,)).fetchone():
|
||
raise HTTPException(404, "Kanban-Projekt nicht gefunden.")
|
||
values = {}
|
||
if "name" in data:
|
||
values["name"] = _name(data["name"])
|
||
if "description" in data:
|
||
values["description"] = _description(data["description"])
|
||
if "completed" in data:
|
||
if not isinstance(data["completed"], bool):
|
||
raise ValidationError("Ungültiger Wert für abgeschlossen.")
|
||
values["completed_at"] = now_iso() if data["completed"] else None
|
||
if "links" in data:
|
||
_save_links(conn, project_id, _links(conn, data["links"]))
|
||
values.update(updated_at=now_iso(), updated_by=user["id"])
|
||
columns = ", ".join(f"{k} = ?" for k in values) # Schlüssel stammen nur aus der Liste oben
|
||
conn.execute(f"UPDATE kanban_projects SET {columns} WHERE id = ?", (*values.values(), project_id))
|
||
return _board(conn, project_id)
|
||
|
||
|
||
@router.delete("/api/kanban/projekte/{project_id}")
|
||
def delete_project(project_id: int, user: dict = Depends(current_user)):
|
||
with transaction() as conn:
|
||
row = conn.execute("SELECT * FROM kanban_projects WHERE id = ?", (project_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "Kanban-Projekt nicht gefunden.")
|
||
if not _may_delete(user, row["created_by"]):
|
||
raise HTTPException(403, "Nur wer das Projekt angelegt hat oder ein Administrator darf es löschen.")
|
||
conn.execute("DELETE FROM kanban_projects WHERE id = ?", (project_id,)) # Aufgaben & Links per CASCADE
|
||
return _overview(conn)
|
||
|
||
|
||
# ---------- Aufgaben ----------
|
||
|
||
|
||
def _get_task(conn, task_id: int):
|
||
row = conn.execute("SELECT * FROM kanban_tasks WHERE id = ?", (task_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "Aufgabe nicht gefunden – vielleicht wurde sie gerade gelöscht.")
|
||
return row
|
||
|
||
|
||
def _end_of(conn, project_id: int, status: str) -> float:
|
||
return _next(conn, "kanban_tasks", "WHERE project_id = ? AND status = ?", (project_id, status))
|
||
|
||
|
||
@router.post("/api/kanban/projekte/{project_id}/aufgaben", status_code=201)
|
||
def create_task(project_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
title = _name(data.get("title"), "Titel", MAX_TITLE)
|
||
description = _description(data.get("description"))
|
||
status = _status(data.get("status", "offen"))
|
||
priority = _priority(data.get("priority", "mittel"))
|
||
with transaction() as conn:
|
||
if not conn.execute("SELECT 1 FROM kanban_projects WHERE id = ?", (project_id,)).fetchone():
|
||
raise HTTPException(404, "Kanban-Projekt nicht gefunden.")
|
||
if conn.execute("SELECT COUNT(*) FROM kanban_tasks WHERE project_id = ?", (project_id,)).fetchone()[0] >= MAX_TASKS:
|
||
raise ValidationError(f"Maximal {MAX_TASKS} Aufgaben pro Projekt. Bitte erledigte Aufgaben löschen.")
|
||
now = now_iso()
|
||
conn.execute(
|
||
"""INSERT INTO kanban_tasks (project_id, title, description, status, assignee, priority, position,
|
||
created_at, created_by, updated_at, updated_by)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
||
(project_id, title, description, status, _assignee(conn, data.get("assignee")), priority,
|
||
_end_of(conn, project_id, status), now, user["id"], now, user["id"]),
|
||
)
|
||
return _board(conn, project_id)
|
||
|
||
|
||
@router.patch("/api/kanban/aufgaben/{task_id}")
|
||
def update_task(task_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
"""Felder einzeln ändern. Neuer Status oder neue Priorität ohne Position: ans Ende der Stufe."""
|
||
with transaction() as conn:
|
||
row = _get_task(conn, task_id)
|
||
values = {}
|
||
if "title" in data:
|
||
values["title"] = _name(data["title"], "Titel", MAX_TITLE)
|
||
if "description" in data:
|
||
values["description"] = _description(data["description"])
|
||
if "assignee" in data:
|
||
values["assignee"] = _assignee(conn, data["assignee"])
|
||
if "status" in data:
|
||
values["status"] = _status(data["status"])
|
||
if "priority" in data:
|
||
values["priority"] = _priority(data["priority"])
|
||
if "position" in data:
|
||
values["position"] = _position(data["position"])
|
||
elif values.get("status", row["status"]) != row["status"] or values.get("priority", row["priority"]) != row["priority"]:
|
||
# Neue Spalte oder neue Priorität: ans Ende (innerhalb der Stufe, da nach Priorität sortiert wird)
|
||
values["position"] = _end_of(conn, row["project_id"], values.get("status", row["status"]))
|
||
if values:
|
||
values.update(updated_at=now_iso(), updated_by=user["id"])
|
||
columns = ", ".join(f"{k} = ?" for k in values) # Schlüssel stammen nur aus der Liste oben
|
||
conn.execute(f"UPDATE kanban_tasks SET {columns} WHERE id = ?", (*values.values(), task_id))
|
||
return _board(conn, row["project_id"])
|
||
|
||
|
||
@router.delete("/api/kanban/aufgaben/{task_id}")
|
||
def delete_task(task_id: int, user: dict = Depends(current_user)):
|
||
with transaction() as conn:
|
||
row = _get_task(conn, task_id)
|
||
if not _may_delete(user, row["created_by"]):
|
||
raise HTTPException(403, "Nur wer die Aufgabe angelegt hat oder ein Administrator darf sie löschen.")
|
||
conn.execute("DELETE FROM kanban_tasks WHERE id = ?", (task_id,))
|
||
return _board(conn, row["project_id"])
|
||
|
||
|
||
# ---------- Für die Artikelseite ----------
|
||
|
||
|
||
def boards_for_article(conn, article_id: str) -> list:
|
||
rows = conn.execute(
|
||
"""SELECT k.id, k.name FROM kanban_project_links l JOIN kanban_projects k ON k.id = l.project_id
|
||
WHERE l.article_id = ? ORDER BY k.name COLLATE NOCASE""",
|
||
(article_id,),
|
||
).fetchall()
|
||
return [{"id": r["id"], "name": r["name"]} for r in rows]
|