feat(obligations): Go PARTIAL DEPRECATED, Python x-user-id, UCCA Proxy Headers, 62 Tests
All checks were successful
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / test-go-ai-compliance (push) Successful in 31s
CI / test-python-backend-compliance (push) Successful in 29s
CI / test-python-document-crawler (push) Successful in 19s
CI / test-python-dsms-gateway (push) Successful in 26s

- Go obligations_handlers.go: CRUD-Overlap als deprecated markiert, AI-Features (Assess/Gap/TOM/Export) bleiben aktiv
- Python obligation_routes.py: x-user-id Header + Audit-Logging an 4 Write-Endpoints
- 3 UCCA Proxy-Dateien: Default X-Tenant-ID + X-User-ID Headers
- Tests von 39 auf 62 erweitert (+23 Route-Integration-Tests mit mock_db/TestClient)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-06 21:31:14 +01:00
parent a5e4801b09
commit 3467bce222
6 changed files with 355 additions and 13 deletions

View File

@@ -188,9 +188,11 @@ async def create_obligation(
payload: ObligationCreate,
db: Session = Depends(get_db),
x_tenant_id: Optional[str] = Header(None),
x_user_id: Optional[str] = Header(None),
):
"""Create a new compliance obligation."""
tenant_id = _get_tenant_id(x_tenant_id)
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 [])
@@ -244,9 +246,11 @@ async def update_obligation(
payload: ObligationUpdate,
db: Session = Depends(get_db),
x_tenant_id: Optional[str] = Header(None),
x_user_id: Optional[str] = Header(None),
):
"""Update an obligation's fields."""
tenant_id = _get_tenant_id(x_tenant_id)
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()}
@@ -282,9 +286,11 @@ async def update_obligation_status(
payload: ObligationStatusUpdate,
db: Session = Depends(get_db),
x_tenant_id: Optional[str] = Header(None),
x_user_id: Optional[str] = Header(None),
):
"""Quick status update for an obligation."""
tenant_id = _get_tenant_id(x_tenant_id)
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)}")
@@ -307,8 +313,10 @@ async def delete_obligation(
obligation_id: str,
db: Session = Depends(get_db),
x_tenant_id: Optional[str] = Header(None),
x_user_id: Optional[str] = Header(None),
):
tenant_id = _get_tenant_id(x_tenant_id)
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