"""RecurringElement service — tenant-scoped CRUD (AP0.9e).""" 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.audit import log_audit from services.initiatives import get_initiative RecurringStatus = Literal["active", "paused", "ended"] RECURRING_STATUSES = frozenset({"active", "paused", "ended"}) _RECURRING_COLUMNS = """ id, tenant_id, initiative_id, title, description, status, interval_days, next_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", "initiative_id"): if result.get(key): result[key] = str(result[key]) for ts_key in ("created_at", "updated_at", "next_due_at"): if result.get(ts_key): result[ts_key] = result[ts_key].isoformat() return result def _validate_status(status: str) -> None: if status not in RECURRING_STATUSES: raise ValueError(f"Ungültiger Recurring-Status: {status}") def create_recurring_element( *, tenant_id: str, initiative_id: str, title: str, description: str = "", status: RecurringStatus = "active", interval_days: Optional[int] = None, next_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) if interval_days is not None and interval_days <= 0: raise ValueError("interval_days muss positiv sein") if not get_initiative(tenant_id=tenant_id, initiative_id=initiative_id): raise ValueError("Initiative nicht gefunden") conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( f""" INSERT INTO recurring_elements ( tenant_id, initiative_id, title, description, status, interval_days, next_due_at ) VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING {_RECURRING_COLUMNS} """, ( tenant_id, initiative_id, title, description, status, interval_days, next_due_at, ), ) row = _serialize_row(dict(cur.fetchone())) conn.commit() finally: conn.close() log_audit( "recurring.created", user_id=user_id, tenant_id=tenant_id, details={"recurring_id": row["id"], "initiative_id": initiative_id}, ) return row def list_recurring_for_initiative( *, tenant_id: str, initiative_id: str ) -> list[dict[str, Any]]: if not get_initiative(tenant_id=tenant_id, initiative_id=initiative_id): raise ValueError("Initiative nicht gefunden") conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( f""" SELECT {_RECURRING_COLUMNS} FROM recurring_elements WHERE tenant_id = %s AND initiative_id = %s ORDER BY updated_at DESC, title """, (tenant_id, initiative_id), ) return [_serialize_row(dict(r)) for r in cur.fetchall()] finally: conn.close() def get_recurring_element( *, tenant_id: str, recurring_id: str ) -> Optional[dict[str, Any]]: conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( f""" SELECT {_RECURRING_COLUMNS} FROM recurring_elements WHERE id = %s AND tenant_id = %s """, (recurring_id, tenant_id), ) row = cur.fetchone() return _serialize_row(dict(row)) if row else None finally: conn.close() def update_recurring_element( *, tenant_id: str, recurring_id: str, user_id: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, status: Optional[RecurringStatus] = None, interval_days: Optional[int] = None, clear_interval_days: bool = False, next_due_at: Optional[datetime] = None, clear_next_due_at: bool = False, ) -> Optional[dict[str, Any]]: existing = get_recurring_element(tenant_id=tenant_id, recurring_id=recurring_id) if not existing: return None old_status = existing["status"] 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) updates.append("status = %s") params.append(status) if clear_interval_days: updates.append("interval_days = NULL") elif interval_days is not None: if interval_days <= 0: raise ValueError("interval_days muss positiv sein") updates.append("interval_days = %s") params.append(interval_days) if clear_next_due_at: updates.append("next_due_at = NULL") elif next_due_at is not None: updates.append("next_due_at = %s") params.append(next_due_at) if not updates: return existing updates.append("updated_at = NOW()") params.extend([recurring_id, tenant_id]) conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( f""" UPDATE recurring_elements SET {", ".join(updates)} WHERE id = %s AND tenant_id = %s RETURNING {_RECURRING_COLUMNS} """, params, ) row = cur.fetchone() if not row: return None result = _serialize_row(dict(row)) conn.commit() finally: conn.close() log_audit( "recurring.updated", user_id=user_id, tenant_id=tenant_id, details={"recurring_id": recurring_id}, ) if status is not None and status != old_status: log_audit( "recurring.status_changed", user_id=user_id, tenant_id=tenant_id, details={ "recurring_id": recurring_id, "from_status": old_status, "to_status": status, }, ) return result def delete_recurring_element( *, tenant_id: str, recurring_id: str, user_id: Optional[str] = None, ) -> bool: conn = get_connection() try: with conn.cursor() as cur: cur.execute( """ DELETE FROM recurring_elements WHERE id = %s AND tenant_id = %s RETURNING id """, (recurring_id, tenant_id), ) deleted = cur.fetchone() is not None conn.commit() finally: conn.close() if deleted: log_audit( "recurring.deleted", user_id=user_id, tenant_id=tenant_id, details={"recurring_id": recurring_id}, ) return deleted