"""Task service — kleinste Einheit unter Action (AP1.5, AP1.5d rekursiv).""" from __future__ import annotations from datetime import datetime from typing import Any, Literal, Optional from psycopg2.extras import RealDictCursor from db import get_connection from services.actions import get_action from services.audit import log_audit TaskStatus = Literal["open", "in_progress", "done", "discarded"] TASK_STATUSES = frozenset({"open", "in_progress", "done", "discarded"}) TERMINAL_TASK_STATUSES = frozenset({"done", "discarded"}) MAX_TASK_DEPTH = 10 _TASK_COLUMNS = """ id, tenant_id, action_id, parent_task_id, title, description, status, roadmap_item_id, sort_order, due_at, created_at, updated_at """ def _serialize_row(row: dict[str, Any]) -> dict[str, Any]: result = dict(row) for key in ("id", "tenant_id", "action_id", "parent_task_id", "roadmap_item_id"): if result.get(key): result[key] = str(result[key]) if result.get("due_at"): result["due_at"] = result["due_at"].isoformat() for ts in ("created_at", "updated_at"): if result.get(ts): result[ts] = result[ts].isoformat() return result def _validate_status(status: str) -> None: if status not in TASK_STATUSES: raise ValueError(f"Ungültiger Task-Status: {status}") def _get_action_or_raise(*, tenant_id: str, action_id: str) -> dict[str, Any]: action = get_action(tenant_id=tenant_id, action_id=action_id) if not action: raise ValueError("Arbeitspaket nicht gefunden") return action def _validate_roadmap_item_for_action( cur, *, tenant_id: str, initiative_id: str, roadmap_item_id: Optional[str] ) -> None: if not roadmap_item_id: return cur.execute( """ SELECT 1 FROM roadmap_items ri JOIN roadmaps r ON r.id = ri.roadmap_id AND r.tenant_id = ri.tenant_id WHERE ri.id = %s AND ri.tenant_id = %s AND r.initiative_id = %s """, (roadmap_item_id, tenant_id, initiative_id), ) if not cur.fetchone(): raise ValueError("Gate gehört nicht zum Vorhaben des Arbeitspakets") def _load_action_task_maps( cur, *, tenant_id: str, action_id: str ) -> tuple[dict[str, dict[str, Any]], dict[str, list[str]]]: cur.execute( f""" SELECT {_TASK_COLUMNS} FROM tasks WHERE tenant_id = %s AND action_id = %s """, (tenant_id, action_id), ) rows = [dict(r) for r in cur.fetchall()] by_id = {str(r["id"]): r for r in rows} by_parent: dict[str, list[str]] = {} for row in rows: parent_id = row.get("parent_task_id") if parent_id: by_parent.setdefault(str(parent_id), []).append(str(row["id"])) return by_id, by_parent def _collect_descendant_ids(task_id: str, by_parent: dict[str, list[str]]) -> set[str]: result: set[str] = set() stack = list(by_parent.get(task_id, [])) while stack: child_id = stack.pop() if child_id in result: continue result.add(child_id) stack.extend(by_parent.get(child_id, [])) return result def _task_depth_from_map( task_id: str, by_id: dict[str, dict[str, Any]], memo: dict[str, int], ) -> int: if task_id in memo: return memo[task_id] row = by_id.get(task_id) if not row or not row.get("parent_task_id"): memo[task_id] = 1 return 1 parent_id = str(row["parent_task_id"]) depth = 1 + _task_depth_from_map(parent_id, by_id, memo) memo[task_id] = depth return depth def _subtree_height(task_id: str, by_parent: dict[str, list[str]]) -> int: children = by_parent.get(task_id, []) if not children: return 1 return 1 + max(_subtree_height(child_id, by_parent) for child_id in children) def _validate_parent_task( cur, *, tenant_id: str, action_id: str, task_id: Optional[str], parent_task_id: Optional[str], ) -> None: if not parent_task_id: return if task_id and parent_task_id == task_id: raise ValueError("Task kann nicht sein eigener Parent sein") cur.execute( f""" SELECT {_TASK_COLUMNS} FROM tasks WHERE id = %s AND tenant_id = %s AND action_id = %s """, (parent_task_id, tenant_id, action_id), ) parent = cur.fetchone() if not parent: raise ValueError("Parent-Task gehört nicht zu diesem Arbeitspaket") by_id, by_parent = _load_action_task_maps(cur, tenant_id=tenant_id, action_id=action_id) if task_id: descendants = _collect_descendant_ids(task_id, by_parent) if parent_task_id in descendants: raise ValueError("Zyklus: Parent darf kein Nachfahr sein") memo: dict[str, int] = {} parent_depth = _task_depth_from_map(parent_task_id, by_id, memo) subtree_height = _subtree_height(task_id, by_parent) if task_id else 1 if parent_depth + subtree_height > MAX_TASK_DEPTH: raise ValueError(f"Maximale Task-Tiefe ({MAX_TASK_DEPTH}) würde überschritten") def _has_open_children(cur, *, tenant_id: str, task_id: str) -> bool: cur.execute( """ SELECT 1 FROM tasks WHERE tenant_id = %s AND parent_task_id = %s AND status NOT IN ('done', 'discarded') LIMIT 1 """, (tenant_id, task_id), ) return cur.fetchone() is not None def _rollup_parent_status(cur, *, tenant_id: str, parent_task_id: str) -> None: cur.execute( """ SELECT status FROM tasks WHERE tenant_id = %s AND parent_task_id = %s """, (tenant_id, parent_task_id), ) child_statuses = [row[0] for row in cur.fetchall()] if not child_statuses: return if any(status not in TERMINAL_TASK_STATUSES for status in child_statuses): return cur.execute( """ UPDATE tasks SET status = 'done', updated_at = NOW() WHERE id = %s AND tenant_id = %s AND status NOT IN ('done', 'discarded') RETURNING parent_task_id """, (parent_task_id, tenant_id), ) row = cur.fetchone() if row and row[0]: _rollup_parent_status(cur, tenant_id=tenant_id, parent_task_id=str(row[0])) def create_task( *, tenant_id: str, action_id: str, title: str, description: str = "", status: TaskStatus = "open", parent_task_id: Optional[str] = None, roadmap_item_id: Optional[str] = None, sort_order: int = 0, due_at: Optional[datetime] = None, user_id: Optional[str] = None, ) -> dict[str, Any]: title = title.strip() if not title: raise ValueError("Titel ist erforderlich") _validate_status(status) action = _get_action_or_raise(tenant_id=tenant_id, action_id=action_id) conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: _validate_parent_task( cur, tenant_id=tenant_id, action_id=action_id, task_id=None, parent_task_id=parent_task_id, ) _validate_roadmap_item_for_action( cur, tenant_id=tenant_id, initiative_id=action["initiative_id"], roadmap_item_id=roadmap_item_id, ) cur.execute( f""" INSERT INTO tasks ( tenant_id, action_id, parent_task_id, title, description, status, roadmap_item_id, sort_order, due_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING {_TASK_COLUMNS} """, ( tenant_id, action_id, parent_task_id, title, description, status, roadmap_item_id, sort_order, due_at, ), ) row = _serialize_row(dict(cur.fetchone())) conn.commit() finally: conn.close() log_audit( "task.created", user_id=user_id, tenant_id=tenant_id, details={ "task_id": row["id"], "action_id": action_id, "title": title, "parent_task_id": parent_task_id, }, ) return row def list_tasks_for_action(*, tenant_id: str, action_id: str) -> list[dict[str, Any]]: _get_action_or_raise(tenant_id=tenant_id, action_id=action_id) conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( f""" SELECT {_TASK_COLUMNS} FROM tasks WHERE tenant_id = %s AND action_id = %s ORDER BY sort_order ASC, created_at ASC """, (tenant_id, action_id), ) return [_serialize_row(dict(r)) for r in cur.fetchall()] finally: conn.close() def get_task(*, tenant_id: str, task_id: str) -> Optional[dict[str, Any]]: conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( f"SELECT {_TASK_COLUMNS} FROM tasks WHERE id = %s AND tenant_id = %s", (task_id, tenant_id), ) row = cur.fetchone() return _serialize_row(dict(row)) if row else None finally: conn.close() def update_task( *, tenant_id: str, task_id: str, user_id: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, status: Optional[TaskStatus] = None, parent_task_id: Optional[str] = None, clear_parent_task: bool = False, roadmap_item_id: Optional[str] = None, clear_roadmap_item: bool = False, sort_order: Optional[int] = None, due_at: Optional[datetime] = None, clear_due_at: bool = False, ) -> Optional[dict[str, Any]]: existing = get_task(tenant_id=tenant_id, task_id=task_id) if not existing: return None action = _get_action_or_raise(tenant_id=tenant_id, action_id=existing["action_id"]) updates: list[str] = [] params: list[Any] = [] if title is not None: title = title.strip() if not title: raise ValueError("Titel ist erforderlich") updates.append("title = %s") params.append(title) if description is not None: updates.append("description = %s") params.append(description) if status is not None: _validate_status(status) if clear_parent_task: updates.append("parent_task_id = NULL") elif parent_task_id is not None: updates.append("parent_task_id = %s") params.append(parent_task_id) if clear_roadmap_item: updates.append("roadmap_item_id = NULL") elif roadmap_item_id is not None: updates.append("roadmap_item_id = %s") params.append(roadmap_item_id) if sort_order is not None: updates.append("sort_order = %s") params.append(sort_order) if clear_due_at: updates.append("due_at = NULL") elif due_at is not None: updates.append("due_at = %s") params.append(due_at) conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: if parent_task_id is not None and not clear_parent_task: _validate_parent_task( cur, tenant_id=tenant_id, action_id=existing["action_id"], task_id=task_id, parent_task_id=parent_task_id, ) if roadmap_item_id is not None and not clear_roadmap_item: _validate_roadmap_item_for_action( cur, tenant_id=tenant_id, initiative_id=action["initiative_id"], roadmap_item_id=roadmap_item_id, ) if status is not None: if status == "done" and _has_open_children( cur, tenant_id=tenant_id, task_id=task_id ): raise ValueError( "Task kann nicht abgeschlossen werden — offene Unteraufgaben vorhanden" ) updates.append("status = %s") params.append(status) if not updates: return existing updates.append("updated_at = NOW()") params.extend([task_id, tenant_id]) cur.execute( f""" UPDATE tasks SET {", ".join(updates)} WHERE id = %s AND tenant_id = %s RETURNING {_TASK_COLUMNS} """, params, ) row = cur.fetchone() if not row: return None result = _serialize_row(dict(row)) if status == "done" and result.get("parent_task_id"): _rollup_parent_status( cur, tenant_id=tenant_id, parent_task_id=str(result["parent_task_id"]), ) conn.commit() finally: conn.close() log_audit( "task.updated", user_id=user_id, tenant_id=tenant_id, details={"task_id": task_id}, ) return get_task(tenant_id=tenant_id, task_id=task_id) def delete_task(*, tenant_id: str, task_id: str, user_id: Optional[str] = None) -> bool: conn = get_connection() try: with conn.cursor() as cur: cur.execute( """ SELECT 1 FROM tasks WHERE parent_task_id = %s AND tenant_id = %s LIMIT 1 """, (task_id, tenant_id), ) if cur.fetchone(): raise ValueError( "Task hat Unteraufgaben — zuerst löschen oder verschieben" ) cur.execute( "DELETE FROM tasks WHERE id = %s AND tenant_id = %s RETURNING id", (task_id, tenant_id), ) deleted = cur.fetchone() is not None conn.commit() finally: conn.close() if deleted: log_audit( "task.deleted", user_id=user_id, tenant_id=tenant_id, details={"task_id": task_id}, ) return deleted