Benjamin_Boenisch/breakpilot-compliance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(backend/api): extract VVTService (Step 4 — file 5 of 18)

Browse Source 
compliance/api/vvt_routes.py (550 LOC) -> 225 LOC thin routes + 475-line
VVTService. Covers the organization header, processing activities CRUD,
audit log, JSON/CSV export, stats, and version lookups for the Art. 30
DSGVO Verzeichnis.

Single-service split: organization + activities + audit + stats all
revolve around the same tenant's VVT document, and the existing test
suite (tests/test_vvt_routes.py — 768 LOC, tests/test_vvt_tenant_isolation.py
— 205 LOC) exercises them together.

Module-level helpers (_activity_to_response, _log_audit, _export_csv)
stay module-level in compliance.services.vvt_service and are re-exported
from compliance.api.vvt_routes so the two test files keep importing
from the old path.

Pydantic schemas already live in compliance.schemas.vvt from Step 3 —
no new schema file needed this round.

mypy.ini flips compliance.api.vvt_routes from ignore_errors=True to
False. Two SQLAlchemy Column[str] vs str dict-index errors fixed with
explicit str() casts on status/business_function in the stats loop.

Verified:
  - 242/242 pytest (173 core + 69 VVT integration) pass
  - OpenAPI 360/484 unchanged
  - mypy compliance/ -> Success on 128 source files
  - vvt_routes.py 550 -> 225 LOC
  - vvt_service.py 475 LOC (under 500 hard cap)
  - Hard-cap violations: 14 -> 13

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sharang Parnerkar
2026-04-07 19:50:40 +02:00
parent f39c7ca40c
commit 4fa0dd6f6d
4 changed files with 587 additions and 425 deletions
Download Patch File Download Diff File Expand all files Collapse all files

517
backend-compliance/compliance/api/vvt_routes.py
View File

@@ -2,62 +2,54 @@
FastAPI routes for VVT — Verzeichnis von Verarbeitungstaetigkeiten (Art. 30 DSGVO).
Endpoints:
GET /vvt/organization — Load organization header
PUT /vvt/organization — Save organization header
GET /vvt/activities — List activities (filter: status, business_function)
POST /vvt/activities — Create new activity
GET /vvt/activities/{id} — Get single activity
PUT /vvt/activities/{id} — Update activity
DELETE /vvt/activities/{id} — Delete activity
GET /vvt/audit-log — Audit trail (limit, offset)
GET /vvt/export — JSON export of all activities
GET /vvt/stats — Statistics
GET /vvt/organization — Load organization header
PUT /vvt/organization — Save organization header
GET /vvt/activities — List activities
POST /vvt/activities — Create new activity
GET /vvt/activities/{id} — Get single activity
PUT /vvt/activities/{id} — Update activity
DELETE /vvt/activities/{id} — Delete activity
GET /vvt/audit-log — Audit trail
GET /vvt/export — JSON or CSV export
GET /vvt/stats — Statistics
GET /vvt/activities/{id}/versions — List activity versions
GET /vvt/activities/{id}/versions/{n} — Get specific version
Phase 1 Step 4 refactor: handlers delegate to VVTService.
"""
import csv
import io
import logging
from datetime import datetime, timezone
from typing import Optional, List
from typing import Any, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi import APIRouter, Depends, Query, Request
from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session
from classroom_engine.database import get_db
from ..db.vvt_models import VVTOrganizationDB, VVTActivityDB, VVTAuditLogDB
from .schemas import (
VVTOrganizationUpdate, VVTOrganizationResponse,
VVTActivityCreate, VVTActivityUpdate, VVTActivityResponse,
VVTStatsResponse, VVTAuditLogEntry,
from compliance.api._http_errors import translate_domain_errors
from compliance.api.tenant_utils import get_tenant_id
from compliance.schemas.vvt import (
VVTActivityCreate,
VVTActivityResponse,
VVTActivityUpdate,
VVTAuditLogEntry,
VVTOrganizationResponse,
VVTOrganizationUpdate,
VVTStatsResponse,
)
from compliance.services.vvt_service import (
VVTService,
_activity_to_response, # re-exported for legacy test imports
_export_csv, # re-exported for legacy test imports
_log_audit, # re-exported for legacy test imports
)
from .tenant_utils import get_tenant_id
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/vvt", tags=["compliance-vvt"])
def _log_audit(
db: Session,
tenant_id: str,
action: str,
entity_type: str,
entity_id=None,
changed_by: str = "system",
old_values=None,
new_values=None,
):
entry = VVTAuditLogDB(
tenant_id=tenant_id,
action=action,
entity_type=entity_type,
entity_id=entity_id,
changed_by=changed_by,
old_values=old_values,
new_values=new_values,
)
db.add(entry)
def get_vvt_service(db: Session = Depends(get_db)) -> VVTService:
return VVTService(db)
# ============================================================================
@@ -67,118 +59,28 @@ def _log_audit(
@router.get("/organization", response_model=Optional[VVTOrganizationResponse])
async def get_organization(
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> Optional[VVTOrganizationResponse]:
"""Load the VVT organization header for the given tenant."""
org = (
db.query(VVTOrganizationDB)
.filter(VVTOrganizationDB.tenant_id == tid)
.order_by(VVTOrganizationDB.created_at)
.first()
)
if not org:
return None
return VVTOrganizationResponse(
id=str(org.id),
organization_name=org.organization_name,
industry=org.industry,
locations=org.locations or [],
employee_count=org.employee_count,
dpo_name=org.dpo_name,
dpo_contact=org.dpo_contact,
vvt_version=org.vvt_version or '1.0',
last_review_date=org.last_review_date,
next_review_date=org.next_review_date,
review_interval=org.review_interval or 'annual',
created_at=org.created_at,
updated_at=org.updated_at,
)
with translate_domain_errors():
return service.get_organization(tid)
@router.put("/organization", response_model=VVTOrganizationResponse)
async def upsert_organization(
request: VVTOrganizationUpdate,
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> VVTOrganizationResponse:
"""Create or update the VVT organization header."""
org = (
db.query(VVTOrganizationDB)
.filter(VVTOrganizationDB.tenant_id == tid)
.order_by(VVTOrganizationDB.created_at)
.first()
)
if not org:
data = request.dict(exclude_none=True)
if 'organization_name' not in data:
data['organization_name'] = 'Meine Organisation'
data['tenant_id'] = tid
org = VVTOrganizationDB(**data)
db.add(org)
else:
for field, value in request.dict(exclude_none=True).items():
setattr(org, field, value)
org.updated_at = datetime.now(timezone.utc)
db.commit()
db.refresh(org)
return VVTOrganizationResponse(
id=str(org.id),
organization_name=org.organization_name,
industry=org.industry,
locations=org.locations or [],
employee_count=org.employee_count,
dpo_name=org.dpo_name,
dpo_contact=org.dpo_contact,
vvt_version=org.vvt_version or '1.0',
last_review_date=org.last_review_date,
next_review_date=org.next_review_date,
review_interval=org.review_interval or 'annual',
created_at=org.created_at,
updated_at=org.updated_at,
)
with translate_domain_errors():
return service.upsert_organization(tid, request)
# ============================================================================
# Activities
# ============================================================================
def _activity_to_response(act: VVTActivityDB) -> VVTActivityResponse:
return VVTActivityResponse(
id=str(act.id),
vvt_id=act.vvt_id,
name=act.name,
description=act.description,
purposes=act.purposes or [],
legal_bases=act.legal_bases or [],
data_subject_categories=act.data_subject_categories or [],
personal_data_categories=act.personal_data_categories or [],
recipient_categories=act.recipient_categories or [],
third_country_transfers=act.third_country_transfers or [],
retention_period=act.retention_period or {},
tom_description=act.tom_description,
business_function=act.business_function,
systems=act.systems or [],
deployment_model=act.deployment_model,
data_sources=act.data_sources or [],
data_flows=act.data_flows or [],
protection_level=act.protection_level or 'MEDIUM',
dpia_required=act.dpia_required or False,
structured_toms=act.structured_toms or {},
status=act.status or 'DRAFT',
responsible=act.responsible,
owner=act.owner,
last_reviewed_at=act.last_reviewed_at,
next_review_at=act.next_review_at,
created_by=act.created_by,
dsfa_id=str(act.dsfa_id) if act.dsfa_id else None,
created_at=act.created_at,
updated_at=act.updated_at,
)
@router.get("/activities", response_model=List[VVTActivityResponse])
async def list_activities(
status: Optional[str] = Query(None),
@@ -186,31 +88,13 @@ async def list_activities(
search: Optional[str] = Query(None),
review_overdue: Optional[bool] = Query(None),
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> List[VVTActivityResponse]:
"""List all processing activities with optional filters."""
query = db.query(VVTActivityDB).filter(VVTActivityDB.tenant_id == tid)
if status:
query = query.filter(VVTActivityDB.status == status)
if business_function:
query = query.filter(VVTActivityDB.business_function == business_function)
if review_overdue:
now = datetime.now(timezone.utc)
query = query.filter(
VVTActivityDB.next_review_at.isnot(None),
VVTActivityDB.next_review_at < now,
with translate_domain_errors():
return service.list_activities(
tid, status, business_function, search, review_overdue
)
if search:
term = f"%{search}%"
query = query.filter(
(VVTActivityDB.name.ilike(term)) |
(VVTActivityDB.description.ilike(term)) |
(VVTActivityDB.vvt_id.ilike(term))
)
activities = query.order_by(VVTActivityDB.created_at.desc()).all()
return [_activity_to_response(a) for a in activities]
@router.post("/activities", response_model=VVTActivityResponse, status_code=201)
@@ -218,58 +102,24 @@ async def create_activity(
request: VVTActivityCreate,
http_request: Request,
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> VVTActivityResponse:
"""Create a new processing activity."""
# Check for duplicate vvt_id within tenant
existing = db.query(VVTActivityDB).filter(
VVTActivityDB.tenant_id == tid,
VVTActivityDB.vvt_id == request.vvt_id,
).first()
if existing:
raise HTTPException(
status_code=409,
detail=f"Activity with VVT-ID '{request.vvt_id}' already exists"
with translate_domain_errors():
return service.create_activity(
tid, request, http_request.headers.get("X-User-ID")
)
data = request.dict()
data['tenant_id'] = tid
# Set created_by from X-User-ID header if not provided in body
if not data.get('created_by'):
data['created_by'] = http_request.headers.get('X-User-ID', 'system')
act = VVTActivityDB(**data)
db.add(act)
db.flush() # get ID before audit log
_log_audit(
db,
tenant_id=tid,
action="CREATE",
entity_type="activity",
entity_id=act.id,
new_values={"vvt_id": act.vvt_id, "name": act.name, "status": act.status},
)
db.commit()
db.refresh(act)
return _activity_to_response(act)
@router.get("/activities/{activity_id}", response_model=VVTActivityResponse)
async def get_activity(
activity_id: str,
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> VVTActivityResponse:
"""Get a single processing activity by ID."""
act = db.query(VVTActivityDB).filter(
VVTActivityDB.id == activity_id,
VVTActivityDB.tenant_id == tid,
).first()
if not act:
raise HTTPException(status_code=404, detail=f"Activity {activity_id} not found")
return _activity_to_response(act)
with translate_domain_errors():
return service.get_activity(tid, activity_id)
@router.put("/activities/{activity_id}", response_model=VVTActivityResponse)
@@ -277,63 +127,22 @@ async def update_activity(
activity_id: str,
request: VVTActivityUpdate,
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> VVTActivityResponse:
"""Update a processing activity."""
act = db.query(VVTActivityDB).filter(
VVTActivityDB.id == activity_id,
VVTActivityDB.tenant_id == tid,
).first()
if not act:
raise HTTPException(status_code=404, detail=f"Activity {activity_id} not found")
old_values = {"name": act.name, "status": act.status}
updates = request.dict(exclude_none=True)
for field, value in updates.items():
setattr(act, field, value)
act.updated_at = datetime.now(timezone.utc)
_log_audit(
db,
tenant_id=tid,
action="UPDATE",
entity_type="activity",
entity_id=act.id,
old_values=old_values,
new_values=updates,
)
db.commit()
db.refresh(act)
return _activity_to_response(act)
with translate_domain_errors():
return service.update_activity(tid, activity_id, request)
@router.delete("/activities/{activity_id}")
async def delete_activity(
activity_id: str,
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> dict[str, Any]:
"""Delete a processing activity."""
act = db.query(VVTActivityDB).filter(
VVTActivityDB.id == activity_id,
VVTActivityDB.tenant_id == tid,
).first()
if not act:
raise HTTPException(status_code=404, detail=f"Activity {activity_id} not found")
_log_audit(
db,
tenant_id=tid,
action="DELETE",
entity_type="activity",
entity_id=act.id,
old_values={"vvt_id": act.vvt_id, "name": act.name},
)
db.delete(act)
db.commit()
return {"success": True, "message": f"Activity {activity_id} deleted"}
with translate_domain_errors():
return service.delete_activity(tid, activity_id)
# ============================================================================
@@ -345,30 +154,11 @@ async def get_audit_log(
limit: int = Query(50, ge=1, le=500),
offset: int = Query(0, ge=0),
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> List[VVTAuditLogEntry]:
"""Get the VVT audit trail."""
entries = (
db.query(VVTAuditLogDB)
.filter(VVTAuditLogDB.tenant_id == tid)
.order_by(VVTAuditLogDB.created_at.desc())
.offset(offset)
.limit(limit)
.all()
)
return [
VVTAuditLogEntry(
id=str(e.id),
action=e.action,
entity_type=e.entity_type,
entity_id=str(e.entity_id) if e.entity_id else None,
changed_by=e.changed_by,
old_values=e.old_values,
new_values=e.new_values,
created_at=e.created_at,
)
for e in entries
]
with translate_domain_errors():
return service.audit_log(tid, limit, offset)
# ============================================================================
@@ -379,145 +169,21 @@ async def get_audit_log(
async def export_activities(
format: str = Query("json", pattern="^(json|csv)$"),
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> Any:
"""Export all activities as JSON or CSV (semicolon-separated, DE locale)."""
org = (
db.query(VVTOrganizationDB)
.filter(VVTOrganizationDB.tenant_id == tid)
.order_by(VVTOrganizationDB.created_at)
.first()
)
activities = (
db.query(VVTActivityDB)
.filter(VVTActivityDB.tenant_id == tid)
.order_by(VVTActivityDB.created_at)
.all()
)
_log_audit(
db,
tenant_id=tid,
action="EXPORT",
entity_type="all_activities",
new_values={"count": len(activities), "format": format},
)
db.commit()
if format == "csv":
return _export_csv(activities)
return {
"exported_at": datetime.now(timezone.utc).isoformat(),
"organization": {
"name": org.organization_name if org else "",
"dpo_name": org.dpo_name if org else "",
"dpo_contact": org.dpo_contact if org else "",
"vvt_version": org.vvt_version if org else "1.0",
} if org else None,
"activities": [
{
"id": str(a.id),
"vvt_id": a.vvt_id,
"name": a.name,
"description": a.description,
"status": a.status,
"purposes": a.purposes,
"legal_bases": a.legal_bases,
"data_subject_categories": a.data_subject_categories,
"personal_data_categories": a.personal_data_categories,
"recipient_categories": a.recipient_categories,
"third_country_transfers": a.third_country_transfers,
"retention_period": a.retention_period,
"dpia_required": a.dpia_required,
"protection_level": a.protection_level,
"business_function": a.business_function,
"responsible": a.responsible,
"created_by": a.created_by,
"dsfa_id": str(a.dsfa_id) if a.dsfa_id else None,
"last_reviewed_at": a.last_reviewed_at.isoformat() if a.last_reviewed_at else None,
"next_review_at": a.next_review_at.isoformat() if a.next_review_at else None,
"created_at": a.created_at.isoformat(),
"updated_at": a.updated_at.isoformat() if a.updated_at else None,
}
for a in activities
],
}
def _export_csv(activities: list) -> StreamingResponse:
"""Generate semicolon-separated CSV with UTF-8 BOM for German Excel compatibility."""
output = io.StringIO()
# UTF-8 BOM for Excel
output.write('\ufeff')
writer = csv.writer(output, delimiter=';', quoting=csv.QUOTE_MINIMAL)
writer.writerow([
'ID', 'VVT-ID', 'Name', 'Zweck', 'Rechtsgrundlage',
'Datenkategorien', 'Betroffene', 'Empfaenger', 'Drittland',
'Aufbewahrung', 'Status', 'Verantwortlich', 'Erstellt von',
'Erstellt am',
])
for a in activities:
writer.writerow([
str(a.id),
a.vvt_id,
a.name,
'; '.join(a.purposes or []),
'; '.join(a.legal_bases or []),
'; '.join(a.personal_data_categories or []),
'; '.join(a.data_subject_categories or []),
'; '.join(a.recipient_categories or []),
'Ja' if a.third_country_transfers else 'Nein',
str(a.retention_period) if a.retention_period else '',
a.status or 'DRAFT',
a.responsible or '',
a.created_by or 'system',
a.created_at.strftime('%d.%m.%Y %H:%M') if a.created_at else '',
])
output.seek(0)
return StreamingResponse(
iter([output.getvalue()]),
media_type='text/csv; charset=utf-8',
headers={
'Content-Disposition': f'attachment; filename="vvt_export_{datetime.now(timezone.utc).strftime("%Y%m%d")}.csv"'
},
)
with translate_domain_errors():
return service.export(tid, format)
@router.get("/stats", response_model=VVTStatsResponse)
async def get_stats(
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> VVTStatsResponse:
"""Get VVT statistics summary."""
activities = db.query(VVTActivityDB).filter(VVTActivityDB.tenant_id == tid).all()
by_status: dict = {}
by_bf: dict = {}
now = datetime.now(timezone.utc)
overdue_count = 0
for a in activities:
status = a.status or 'DRAFT'
bf = a.business_function or 'unknown'
by_status[status] = by_status.get(status, 0) + 1
by_bf[bf] = by_bf.get(bf, 0) + 1
if a.next_review_at and a.next_review_at < now:
overdue_count += 1
return VVTStatsResponse(
total=len(activities),
by_status=by_status,
by_business_function=by_bf,
dpia_required_count=sum(1 for a in activities if a.dpia_required),
third_country_count=sum(1 for a in activities if a.third_country_transfers),
draft_count=by_status.get('DRAFT', 0),
approved_count=by_status.get('APPROVED', 0),
overdue_review_count=overdue_count,
)
with translate_domain_errors():
return service.stats(tid)
# ============================================================================
@@ -528,11 +194,11 @@ async def get_stats(
async def list_activity_versions(
activity_id: str,
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> Any:
"""List all versions for a VVT activity."""
from .versioning_utils import list_versions
return list_versions(db, "vvt_activity", activity_id, tid)
with translate_domain_errors():
return service.list_versions(tid, activity_id)
@router.get("/activities/{activity_id}/versions/{version_number}")
@@ -540,11 +206,20 @@ async def get_activity_version(
activity_id: str,
version_number: int,
tid: str = Depends(get_tenant_id),
db: Session = Depends(get_db),
):
service: VVTService = Depends(get_vvt_service),
) -> Any:
"""Get a specific VVT activity version with full snapshot."""
from .versioning_utils import get_version
v = get_version(db, "vvt_activity", activity_id, version_number, tid)
if not v:
raise HTTPException(status_code=404, detail=f"Version {version_number} not found")
return v
with translate_domain_errors():
return service.get_version(tid, activity_id, version_number)
# ----------------------------------------------------------------------------
# Legacy re-exports for tests that import helpers directly.
# ----------------------------------------------------------------------------
__all__ = [
"router",
"_activity_to_response",
"_log_audit",
"_export_csv",
]