c3afa628ed
Integrate the vendor-compliance module with four DSGVO modules to eliminate data silos and resolve the VVT processor tab's ephemeral state problem. - Reposition vendor-compliance sidebar from seq 4200 to 2500 (after VVT) - VVT: replace ephemeral ProcessorRecord state with Vendor-API fetch (read-only) - Obligations: add linked_vendor_ids (JSONB) + compliance check #12 MISSING_VENDOR_LINK - TOM: add vendor TOM-controls cross-reference table in overview tab - Loeschfristen: add linked_vendor_ids (JSONB) + vendor picker + document section - Migrations: 069_obligations_vendor_link.sql, 070_loeschfristen_vendor_link.sql - Tests: 12 new backend tests (125 total pass) - Docs: update obligations.md + vendors.md with cross-module integration Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
337 lines
12 KiB
Python
337 lines
12 KiB
Python
"""
|
|
FastAPI routes for Compliance Obligations Tracking.
|
|
|
|
Endpoints:
|
|
GET /obligations — list with filters (status, priority, source, limit, offset)
|
|
POST /obligations — create obligation
|
|
GET /obligations/stats — counts per status and priority
|
|
GET /obligations/{id} — get single obligation
|
|
PUT /obligations/{id} — update obligation
|
|
PUT /obligations/{id}/status — quick status update
|
|
DELETE /obligations/{id} — delete obligation
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Optional, List, Any, Dict
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Header
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from classroom_engine.database import get_db
|
|
from .tenant_utils import get_tenant_id as _get_tenant_id
|
|
from .db_utils import row_to_dict as _row_to_dict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/obligations", tags=["obligations"])
|
|
|
|
|
|
# =============================================================================
|
|
# Pydantic Schemas
|
|
# =============================================================================
|
|
|
|
class ObligationCreate(BaseModel):
|
|
title: str
|
|
description: Optional[str] = None
|
|
source: str = "DSGVO"
|
|
source_article: Optional[str] = None
|
|
deadline: Optional[datetime] = None
|
|
status: str = "pending"
|
|
priority: str = "medium"
|
|
responsible: Optional[str] = None
|
|
linked_systems: Optional[List[str]] = None
|
|
linked_vendor_ids: Optional[List[str]] = None
|
|
assessment_id: Optional[str] = None
|
|
rule_code: Optional[str] = None
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class ObligationUpdate(BaseModel):
|
|
title: Optional[str] = None
|
|
description: Optional[str] = None
|
|
source: Optional[str] = None
|
|
source_article: Optional[str] = None
|
|
deadline: Optional[datetime] = None
|
|
status: Optional[str] = None
|
|
priority: Optional[str] = None
|
|
responsible: Optional[str] = None
|
|
linked_systems: Optional[List[str]] = None
|
|
linked_vendor_ids: Optional[List[str]] = None
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class ObligationStatusUpdate(BaseModel):
|
|
status: str
|
|
|
|
|
|
|
|
# =============================================================================
|
|
# Routes
|
|
# =============================================================================
|
|
|
|
@router.get("")
|
|
async def list_obligations(
|
|
status: Optional[str] = Query(None),
|
|
priority: Optional[str] = Query(None),
|
|
source: Optional[str] = Query(None),
|
|
search: Optional[str] = Query(None),
|
|
limit: int = Query(100, ge=1, le=500),
|
|
offset: int = Query(0, ge=0),
|
|
db: Session = Depends(get_db),
|
|
tenant_id: str = Depends(_get_tenant_id),
|
|
):
|
|
"""List obligations with optional filters."""
|
|
|
|
where_clauses = ["tenant_id = :tenant_id"]
|
|
params: Dict[str, Any] = {"tenant_id": tenant_id, "limit": limit, "offset": offset}
|
|
|
|
if status:
|
|
where_clauses.append("status = :status")
|
|
params["status"] = status
|
|
if priority:
|
|
where_clauses.append("priority = :priority")
|
|
params["priority"] = priority
|
|
if source:
|
|
where_clauses.append("source ILIKE :source")
|
|
params["source"] = f"%{source}%"
|
|
if search:
|
|
where_clauses.append("(title ILIKE :search OR description ILIKE :search)")
|
|
params["search"] = f"%{search}%"
|
|
|
|
where_sql = " AND ".join(where_clauses)
|
|
|
|
total_row = db.execute(
|
|
text(f"SELECT COUNT(*) FROM compliance_obligations WHERE {where_sql}"),
|
|
params,
|
|
).fetchone()
|
|
total = total_row[0] if total_row else 0
|
|
|
|
rows = db.execute(
|
|
text(f"""
|
|
SELECT * FROM compliance_obligations
|
|
WHERE {where_sql}
|
|
ORDER BY
|
|
CASE priority
|
|
WHEN 'critical' THEN 0
|
|
WHEN 'high' THEN 1
|
|
WHEN 'medium' THEN 2
|
|
ELSE 3
|
|
END,
|
|
CASE status
|
|
WHEN 'overdue' THEN 0
|
|
WHEN 'in-progress' THEN 1
|
|
WHEN 'pending' THEN 2
|
|
ELSE 3
|
|
END,
|
|
created_at DESC
|
|
LIMIT :limit OFFSET :offset
|
|
"""),
|
|
params,
|
|
).fetchall()
|
|
|
|
return {
|
|
"obligations": [_row_to_dict(r) for r in rows],
|
|
"total": total,
|
|
}
|
|
|
|
|
|
@router.get("/stats")
|
|
async def get_obligation_stats(
|
|
db: Session = Depends(get_db),
|
|
tenant_id: str = Depends(_get_tenant_id),
|
|
):
|
|
"""Return obligation counts per status and priority."""
|
|
|
|
rows = db.execute(text("""
|
|
SELECT
|
|
COUNT(*) FILTER (WHERE status = 'pending') AS pending,
|
|
COUNT(*) FILTER (WHERE status = 'in-progress') AS in_progress,
|
|
COUNT(*) FILTER (WHERE status = 'overdue') AS overdue,
|
|
COUNT(*) FILTER (WHERE status = 'completed') AS completed,
|
|
COUNT(*) FILTER (WHERE priority = 'critical') AS critical,
|
|
COUNT(*) FILTER (WHERE priority = 'high') AS high,
|
|
COUNT(*) AS total
|
|
FROM compliance_obligations
|
|
WHERE tenant_id = :tenant_id
|
|
"""), {"tenant_id": tenant_id}).fetchone()
|
|
|
|
if rows:
|
|
d = dict(rows._mapping)
|
|
return {k: (v or 0) for k, v in d.items()}
|
|
return {"pending": 0, "in_progress": 0, "overdue": 0, "completed": 0, "critical": 0, "high": 0, "total": 0}
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_obligation(
|
|
payload: ObligationCreate,
|
|
db: Session = Depends(get_db),
|
|
tenant_id: str = Depends(_get_tenant_id),
|
|
x_user_id: Optional[str] = Header(None),
|
|
):
|
|
"""Create a new compliance obligation."""
|
|
logger.info("create_obligation user_id=%s tenant_id=%s title=%s", x_user_id, tenant_id, payload.title)
|
|
|
|
import json
|
|
linked_systems = json.dumps(payload.linked_systems or [])
|
|
linked_vendor_ids = json.dumps(payload.linked_vendor_ids or [])
|
|
|
|
row = db.execute(text("""
|
|
INSERT INTO compliance_obligations
|
|
(tenant_id, title, description, source, source_article, deadline,
|
|
status, priority, responsible, linked_systems, linked_vendor_ids, assessment_id, rule_code, notes)
|
|
VALUES
|
|
(:tenant_id, :title, :description, :source, :source_article, :deadline,
|
|
:status, :priority, :responsible, CAST(:linked_systems AS jsonb), CAST(:linked_vendor_ids AS jsonb), :assessment_id, :rule_code, :notes)
|
|
RETURNING *
|
|
"""), {
|
|
"tenant_id": tenant_id,
|
|
"title": payload.title,
|
|
"description": payload.description,
|
|
"source": payload.source,
|
|
"source_article": payload.source_article,
|
|
"deadline": payload.deadline,
|
|
"status": payload.status,
|
|
"priority": payload.priority,
|
|
"responsible": payload.responsible,
|
|
"linked_systems": linked_systems,
|
|
"linked_vendor_ids": linked_vendor_ids,
|
|
"assessment_id": payload.assessment_id,
|
|
"rule_code": payload.rule_code,
|
|
"notes": payload.notes,
|
|
}).fetchone()
|
|
db.commit()
|
|
return _row_to_dict(row)
|
|
|
|
|
|
@router.get("/{obligation_id}")
|
|
async def get_obligation(
|
|
obligation_id: str,
|
|
db: Session = Depends(get_db),
|
|
tenant_id: str = Depends(_get_tenant_id),
|
|
):
|
|
row = db.execute(text("""
|
|
SELECT * FROM compliance_obligations
|
|
WHERE id = :id AND tenant_id = :tenant_id
|
|
"""), {"id": obligation_id, "tenant_id": tenant_id}).fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Obligation not found")
|
|
return _row_to_dict(row)
|
|
|
|
|
|
@router.put("/{obligation_id}")
|
|
async def update_obligation(
|
|
obligation_id: str,
|
|
payload: ObligationUpdate,
|
|
db: Session = Depends(get_db),
|
|
tenant_id: str = Depends(_get_tenant_id),
|
|
x_user_id: Optional[str] = Header(None),
|
|
):
|
|
"""Update an obligation's fields."""
|
|
logger.info("update_obligation user_id=%s tenant_id=%s id=%s", x_user_id, tenant_id, obligation_id)
|
|
import json
|
|
|
|
updates: Dict[str, Any] = {"id": obligation_id, "tenant_id": tenant_id, "updated_at": datetime.utcnow()}
|
|
set_clauses = ["updated_at = :updated_at"]
|
|
|
|
for field, value in payload.model_dump(exclude_unset=True).items():
|
|
if field == "linked_systems":
|
|
updates["linked_systems"] = json.dumps(value or [])
|
|
set_clauses.append("linked_systems = CAST(:linked_systems AS jsonb)")
|
|
elif field == "linked_vendor_ids":
|
|
updates["linked_vendor_ids"] = json.dumps(value or [])
|
|
set_clauses.append("linked_vendor_ids = CAST(:linked_vendor_ids AS jsonb)")
|
|
else:
|
|
updates[field] = value
|
|
set_clauses.append(f"{field} = :{field}")
|
|
|
|
if len(set_clauses) == 1:
|
|
raise HTTPException(status_code=400, detail="No fields to update")
|
|
|
|
row = db.execute(text(f"""
|
|
UPDATE compliance_obligations
|
|
SET {', '.join(set_clauses)}
|
|
WHERE id = :id AND tenant_id = :tenant_id
|
|
RETURNING *
|
|
"""), updates).fetchone()
|
|
db.commit()
|
|
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Obligation not found")
|
|
return _row_to_dict(row)
|
|
|
|
|
|
@router.put("/{obligation_id}/status")
|
|
async def update_obligation_status(
|
|
obligation_id: str,
|
|
payload: ObligationStatusUpdate,
|
|
db: Session = Depends(get_db),
|
|
tenant_id: str = Depends(_get_tenant_id),
|
|
x_user_id: Optional[str] = Header(None),
|
|
):
|
|
"""Quick status update for an obligation."""
|
|
logger.info("update_obligation_status user_id=%s tenant_id=%s id=%s status=%s", x_user_id, tenant_id, obligation_id, payload.status)
|
|
valid_statuses = {"pending", "in-progress", "completed", "overdue"}
|
|
if payload.status not in valid_statuses:
|
|
raise HTTPException(status_code=400, detail=f"Invalid status. Must be one of: {', '.join(valid_statuses)}")
|
|
|
|
row = db.execute(text("""
|
|
UPDATE compliance_obligations
|
|
SET status = :status, updated_at = :now
|
|
WHERE id = :id AND tenant_id = :tenant_id
|
|
RETURNING *
|
|
"""), {"status": payload.status, "now": datetime.utcnow(), "id": obligation_id, "tenant_id": tenant_id}).fetchone()
|
|
db.commit()
|
|
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Obligation not found")
|
|
return _row_to_dict(row)
|
|
|
|
|
|
@router.delete("/{obligation_id}", status_code=204)
|
|
async def delete_obligation(
|
|
obligation_id: str,
|
|
db: Session = Depends(get_db),
|
|
tenant_id: str = Depends(_get_tenant_id),
|
|
x_user_id: Optional[str] = Header(None),
|
|
):
|
|
logger.info("delete_obligation user_id=%s tenant_id=%s id=%s", x_user_id, tenant_id, obligation_id)
|
|
result = db.execute(text("""
|
|
DELETE FROM compliance_obligations
|
|
WHERE id = :id AND tenant_id = :tenant_id
|
|
"""), {"id": obligation_id, "tenant_id": tenant_id})
|
|
db.commit()
|
|
if result.rowcount == 0:
|
|
raise HTTPException(status_code=404, detail="Obligation not found")
|
|
|
|
|
|
# =============================================================================
|
|
# Versioning
|
|
# =============================================================================
|
|
|
|
@router.get("/{obligation_id}/versions")
|
|
async def list_obligation_versions(
|
|
obligation_id: str,
|
|
db: Session = Depends(get_db),
|
|
tenant_id: str = Depends(_get_tenant_id),
|
|
):
|
|
"""List all versions for an Obligation."""
|
|
from .versioning_utils import list_versions
|
|
return list_versions(db, "obligation", obligation_id, tenant_id)
|
|
|
|
|
|
@router.get("/{obligation_id}/versions/{version_number}")
|
|
async def get_obligation_version(
|
|
obligation_id: str,
|
|
version_number: int,
|
|
db: Session = Depends(get_db),
|
|
tenant_id: str = Depends(_get_tenant_id),
|
|
):
|
|
"""Get a specific Obligation version with full snapshot."""
|
|
from .versioning_utils import get_version
|
|
v = get_version(db, "obligation", obligation_id, version_number, tenant_id)
|
|
if not v:
|
|
raise HTTPException(status_code=404, detail=f"Version {version_number} not found")
|
|
return v
|