refactor(backend/api): extract DSR services (Step 4 — file 15 of 18)
compliance/api/dsr_routes.py (1176 LOC) -> 369 LOC thin routes +
469-line DsrService + 487-line DsrWorkflowService + 101-line schemas.
Two-service split for Data Subject Request (DSGVO Art. 15-22):
- dsr_service.py: CRUD, list, stats, export, audit log
- dsr_workflow_service.py: identity verification, processing,
portability, escalation
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sharang Parnerkar
487
backend-compliance/compliance/services/dsr_workflow_service.py
Normal file
487
backend-compliance/compliance/services/dsr_workflow_service.py
Normal file
@@ -0,0 +1,487 @@
|
||||
# mypy: disable-error-code="arg-type,assignment,union-attr,no-any-return"
|
||||
"""
|
||||
DSR workflow service — status changes, identity verification, assignment,
|
||||
completion, rejection, communications, exception checks, and templates.
|
||||
|
||||
Phase 1 Step 4: extracted from ``compliance.api.dsr_routes``. CRUD, stats,
|
||||
export, and deadline processing live in ``compliance.services.dsr_service``.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from sqlalchemy import or_
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from compliance.db.dsr_models import (
|
||||
DSRCommunicationDB,
|
||||
DSRExceptionCheckDB,
|
||||
DSRRequestDB,
|
||||
DSRTemplateDB,
|
||||
DSRTemplateVersionDB,
|
||||
)
|
||||
from compliance.domain import NotFoundError, ValidationError
|
||||
from compliance.schemas.dsr import (
|
||||
AssignRequest,
|
||||
CompleteDSR,
|
||||
CreateTemplateVersion,
|
||||
ExtendDeadline,
|
||||
RejectDSR,
|
||||
SendCommunication,
|
||||
StatusChange,
|
||||
UpdateExceptionCheck,
|
||||
VerifyIdentity,
|
||||
)
|
||||
from compliance.services.dsr_service import (
|
||||
ART17_EXCEPTIONS,
|
||||
VALID_STATUSES,
|
||||
_dsr_to_dict,
|
||||
_get_dsr_or_404,
|
||||
_record_history,
|
||||
)
|
||||
|
||||
|
||||
class DSRWorkflowService:
|
||||
"""Workflow actions for DSRs: status, identity, assign, complete, reject,
|
||||
communications, exception checks, templates."""
|
||||
|
||||
def __init__(self, db: Session) -> None:
|
||||
self._db = db
|
||||
|
||||
# -- Status change -------------------------------------------------------
|
||||
|
||||
def change_status(
|
||||
self, dsr_id: str, body: StatusChange, tenant_id: str,
|
||||
) -> Dict[str, Any]:
|
||||
if body.status not in VALID_STATUSES:
|
||||
raise ValidationError(f"Invalid status: {body.status}")
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
_record_history(self._db, dsr, body.status, comment=body.comment)
|
||||
dsr.status = body.status
|
||||
dsr.updated_at = datetime.now(timezone.utc)
|
||||
self._db.commit()
|
||||
self._db.refresh(dsr)
|
||||
return _dsr_to_dict(dsr)
|
||||
|
||||
# -- Identity verification -----------------------------------------------
|
||||
|
||||
def verify_identity(
|
||||
self, dsr_id: str, body: VerifyIdentity, tenant_id: str,
|
||||
) -> Dict[str, Any]:
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
now = datetime.now(timezone.utc)
|
||||
dsr.identity_verified = True
|
||||
dsr.verification_method = body.method
|
||||
dsr.verified_at = now
|
||||
dsr.verified_by = "admin"
|
||||
dsr.verification_notes = body.notes
|
||||
dsr.verification_document_ref = body.document_ref
|
||||
|
||||
if dsr.status == "identity_verification":
|
||||
_record_history(self._db, dsr, "processing", comment="Identitaet verifiziert")
|
||||
dsr.status = "processing"
|
||||
elif dsr.status == "intake":
|
||||
_record_history(self._db, dsr, "identity_verification", comment="Identitaet verifiziert")
|
||||
dsr.status = "identity_verification"
|
||||
|
||||
dsr.updated_at = now
|
||||
self._db.commit()
|
||||
self._db.refresh(dsr)
|
||||
return _dsr_to_dict(dsr)
|
||||
|
||||
# -- Assignment ----------------------------------------------------------
|
||||
|
||||
def assign(
|
||||
self, dsr_id: str, body: AssignRequest, tenant_id: str,
|
||||
) -> Dict[str, Any]:
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
dsr.assigned_to = body.assignee_id
|
||||
dsr.assigned_at = datetime.now(timezone.utc)
|
||||
dsr.assigned_by = "admin"
|
||||
dsr.updated_at = datetime.now(timezone.utc)
|
||||
self._db.commit()
|
||||
self._db.refresh(dsr)
|
||||
return _dsr_to_dict(dsr)
|
||||
|
||||
# -- Extend deadline -----------------------------------------------------
|
||||
|
||||
def extend_deadline(
|
||||
self, dsr_id: str, body: ExtendDeadline, tenant_id: str,
|
||||
) -> Dict[str, Any]:
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
if dsr.status in ("completed", "rejected", "cancelled"):
|
||||
raise ValidationError("Cannot extend deadline for closed DSR")
|
||||
now = datetime.now(timezone.utc)
|
||||
from datetime import timedelta
|
||||
current_deadline = dsr.extended_deadline_at or dsr.deadline_at
|
||||
new_deadline = current_deadline + timedelta(days=body.days or 60)
|
||||
dsr.extended_deadline_at = new_deadline
|
||||
dsr.extension_reason = body.reason
|
||||
dsr.extension_approved_by = "admin"
|
||||
dsr.extension_approved_at = now
|
||||
dsr.updated_at = now
|
||||
_record_history(self._db, dsr, dsr.status, comment=f"Frist verlaengert: {body.reason}")
|
||||
self._db.commit()
|
||||
self._db.refresh(dsr)
|
||||
return _dsr_to_dict(dsr)
|
||||
|
||||
# -- Complete ------------------------------------------------------------
|
||||
|
||||
def complete(
|
||||
self, dsr_id: str, body: CompleteDSR, tenant_id: str,
|
||||
) -> Dict[str, Any]:
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
if dsr.status in ("completed", "cancelled"):
|
||||
raise ValidationError("DSR already completed or cancelled")
|
||||
now = datetime.now(timezone.utc)
|
||||
_record_history(self._db, dsr, "completed", comment=body.summary)
|
||||
dsr.status = "completed"
|
||||
dsr.completed_at = now
|
||||
dsr.completion_notes = body.summary
|
||||
if body.result_data:
|
||||
dsr.data_export = body.result_data
|
||||
dsr.updated_at = now
|
||||
self._db.commit()
|
||||
self._db.refresh(dsr)
|
||||
return _dsr_to_dict(dsr)
|
||||
|
||||
# -- Reject --------------------------------------------------------------
|
||||
|
||||
def reject(
|
||||
self, dsr_id: str, body: RejectDSR, tenant_id: str,
|
||||
) -> Dict[str, Any]:
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
if dsr.status in ("completed", "rejected", "cancelled"):
|
||||
raise ValidationError("DSR already closed")
|
||||
now = datetime.now(timezone.utc)
|
||||
_record_history(self._db, dsr, "rejected", comment=f"{body.reason} ({body.legal_basis})")
|
||||
dsr.status = "rejected"
|
||||
dsr.rejection_reason = body.reason
|
||||
dsr.rejection_legal_basis = body.legal_basis
|
||||
dsr.completed_at = now
|
||||
dsr.updated_at = now
|
||||
self._db.commit()
|
||||
self._db.refresh(dsr)
|
||||
return _dsr_to_dict(dsr)
|
||||
|
||||
# -- History -------------------------------------------------------------
|
||||
|
||||
def get_history(self, dsr_id: str, tenant_id: str) -> List[Dict[str, Any]]:
|
||||
from compliance.db.dsr_models import DSRStatusHistoryDB
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
entries = self._db.query(DSRStatusHistoryDB).filter(
|
||||
DSRStatusHistoryDB.dsr_id == dsr.id,
|
||||
).order_by(DSRStatusHistoryDB.created_at.desc()).all()
|
||||
return [
|
||||
{
|
||||
"id": str(e.id),
|
||||
"dsr_id": str(e.dsr_id),
|
||||
"previous_status": e.previous_status,
|
||||
"new_status": e.new_status,
|
||||
"changed_by": e.changed_by,
|
||||
"comment": e.comment,
|
||||
"created_at": e.created_at.isoformat() if e.created_at else None,
|
||||
}
|
||||
for e in entries
|
||||
]
|
||||
|
||||
# -- Communications ------------------------------------------------------
|
||||
|
||||
def get_communications(self, dsr_id: str, tenant_id: str) -> List[Dict[str, Any]]:
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
comms = self._db.query(DSRCommunicationDB).filter(
|
||||
DSRCommunicationDB.dsr_id == dsr.id,
|
||||
).order_by(DSRCommunicationDB.created_at.desc()).all()
|
||||
return [
|
||||
{
|
||||
"id": str(c.id),
|
||||
"dsr_id": str(c.dsr_id),
|
||||
"communication_type": c.communication_type,
|
||||
"channel": c.channel,
|
||||
"subject": c.subject,
|
||||
"content": c.content,
|
||||
"template_used": c.template_used,
|
||||
"attachments": c.attachments or [],
|
||||
"sent_at": c.sent_at.isoformat() if c.sent_at else None,
|
||||
"sent_by": c.sent_by,
|
||||
"received_at": c.received_at.isoformat() if c.received_at else None,
|
||||
"created_at": c.created_at.isoformat() if c.created_at else None,
|
||||
"created_by": c.created_by,
|
||||
}
|
||||
for c in comms
|
||||
]
|
||||
|
||||
def send_communication(
|
||||
self, dsr_id: str, body: SendCommunication, tenant_id: str,
|
||||
) -> Dict[str, Any]:
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
now = datetime.now(timezone.utc)
|
||||
comm = DSRCommunicationDB(
|
||||
tenant_id=uuid.UUID(tenant_id),
|
||||
dsr_id=dsr.id,
|
||||
communication_type=body.communication_type,
|
||||
channel=body.channel,
|
||||
subject=body.subject,
|
||||
content=body.content,
|
||||
template_used=body.template_used,
|
||||
sent_at=now if body.communication_type == "outgoing" else None,
|
||||
sent_by="admin" if body.communication_type == "outgoing" else None,
|
||||
received_at=now if body.communication_type == "incoming" else None,
|
||||
created_at=now,
|
||||
)
|
||||
self._db.add(comm)
|
||||
self._db.commit()
|
||||
self._db.refresh(comm)
|
||||
return {
|
||||
"id": str(comm.id),
|
||||
"dsr_id": str(comm.dsr_id),
|
||||
"communication_type": comm.communication_type,
|
||||
"channel": comm.channel,
|
||||
"subject": comm.subject,
|
||||
"content": comm.content,
|
||||
"sent_at": comm.sent_at.isoformat() if comm.sent_at else None,
|
||||
"created_at": comm.created_at.isoformat() if comm.created_at else None,
|
||||
}
|
||||
|
||||
# -- Exception checks ----------------------------------------------------
|
||||
|
||||
def get_exception_checks(self, dsr_id: str, tenant_id: str) -> List[Dict[str, Any]]:
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
checks = self._db.query(DSRExceptionCheckDB).filter(
|
||||
DSRExceptionCheckDB.dsr_id == dsr.id,
|
||||
).order_by(DSRExceptionCheckDB.check_code).all()
|
||||
return [
|
||||
{
|
||||
"id": str(c.id),
|
||||
"dsr_id": str(c.dsr_id),
|
||||
"check_code": c.check_code,
|
||||
"article": c.article,
|
||||
"label": c.label,
|
||||
"description": c.description,
|
||||
"applies": c.applies,
|
||||
"notes": c.notes,
|
||||
"checked_by": c.checked_by,
|
||||
"checked_at": c.checked_at.isoformat() if c.checked_at else None,
|
||||
}
|
||||
for c in checks
|
||||
]
|
||||
|
||||
def init_exception_checks(self, dsr_id: str, tenant_id: str) -> List[Dict[str, Any]]:
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
if dsr.request_type != "erasure":
|
||||
raise ValidationError("Exception checks only for erasure requests")
|
||||
existing = self._db.query(DSRExceptionCheckDB).filter(
|
||||
DSRExceptionCheckDB.dsr_id == dsr.id,
|
||||
).count()
|
||||
if existing > 0:
|
||||
raise ValidationError("Exception checks already initialized")
|
||||
|
||||
checks = []
|
||||
for exc in ART17_EXCEPTIONS:
|
||||
check = DSRExceptionCheckDB(
|
||||
tenant_id=uuid.UUID(tenant_id),
|
||||
dsr_id=dsr.id,
|
||||
check_code=exc["check_code"],
|
||||
article=exc["article"],
|
||||
label=exc["label"],
|
||||
description=exc["description"],
|
||||
)
|
||||
self._db.add(check)
|
||||
checks.append(check)
|
||||
self._db.commit()
|
||||
return [
|
||||
{
|
||||
"id": str(c.id),
|
||||
"dsr_id": str(c.dsr_id),
|
||||
"check_code": c.check_code,
|
||||
"article": c.article,
|
||||
"label": c.label,
|
||||
"description": c.description,
|
||||
"applies": c.applies,
|
||||
"notes": c.notes,
|
||||
}
|
||||
for c in checks
|
||||
]
|
||||
|
||||
def update_exception_check(
|
||||
self, dsr_id: str, check_id: str, body: UpdateExceptionCheck, tenant_id: str,
|
||||
) -> Dict[str, Any]:
|
||||
dsr = _get_dsr_or_404(self._db, dsr_id, tenant_id)
|
||||
try:
|
||||
cid = uuid.UUID(check_id)
|
||||
except ValueError:
|
||||
raise ValidationError("Invalid check ID")
|
||||
check = self._db.query(DSRExceptionCheckDB).filter(
|
||||
DSRExceptionCheckDB.id == cid,
|
||||
DSRExceptionCheckDB.dsr_id == dsr.id,
|
||||
).first()
|
||||
if not check:
|
||||
raise NotFoundError("Exception check not found")
|
||||
check.applies = body.applies
|
||||
check.notes = body.notes
|
||||
check.checked_by = "admin"
|
||||
check.checked_at = datetime.now(timezone.utc)
|
||||
self._db.commit()
|
||||
self._db.refresh(check)
|
||||
return {
|
||||
"id": str(check.id),
|
||||
"dsr_id": str(check.dsr_id),
|
||||
"check_code": check.check_code,
|
||||
"article": check.article,
|
||||
"label": check.label,
|
||||
"description": check.description,
|
||||
"applies": check.applies,
|
||||
"notes": check.notes,
|
||||
"checked_by": check.checked_by,
|
||||
"checked_at": check.checked_at.isoformat() if check.checked_at else None,
|
||||
}
|
||||
|
||||
# -- Templates -----------------------------------------------------------
|
||||
|
||||
def get_templates(self, tenant_id: str) -> List[Dict[str, Any]]:
|
||||
templates = self._db.query(DSRTemplateDB).filter(
|
||||
DSRTemplateDB.tenant_id == uuid.UUID(tenant_id),
|
||||
).order_by(DSRTemplateDB.template_type).all()
|
||||
return [
|
||||
{
|
||||
"id": str(t.id),
|
||||
"name": t.name,
|
||||
"template_type": t.template_type,
|
||||
"request_type": t.request_type,
|
||||
"language": t.language,
|
||||
"is_active": t.is_active,
|
||||
"created_at": t.created_at.isoformat() if t.created_at else None,
|
||||
"updated_at": t.updated_at.isoformat() if t.updated_at else None,
|
||||
}
|
||||
for t in templates
|
||||
]
|
||||
|
||||
def get_published_templates(
|
||||
self, tenant_id: str, *, request_type: Optional[str] = None, language: str = "de",
|
||||
) -> List[Dict[str, Any]]:
|
||||
query = self._db.query(DSRTemplateDB).filter(
|
||||
DSRTemplateDB.tenant_id == uuid.UUID(tenant_id),
|
||||
DSRTemplateDB.is_active,
|
||||
DSRTemplateDB.language == language,
|
||||
)
|
||||
if request_type:
|
||||
query = query.filter(
|
||||
or_(
|
||||
DSRTemplateDB.request_type == request_type,
|
||||
DSRTemplateDB.request_type.is_(None),
|
||||
)
|
||||
)
|
||||
templates = query.all()
|
||||
result = []
|
||||
for t in templates:
|
||||
latest = self._db.query(DSRTemplateVersionDB).filter(
|
||||
DSRTemplateVersionDB.template_id == t.id,
|
||||
DSRTemplateVersionDB.status == "published",
|
||||
).order_by(DSRTemplateVersionDB.created_at.desc()).first()
|
||||
result.append({
|
||||
"id": str(t.id),
|
||||
"name": t.name,
|
||||
"template_type": t.template_type,
|
||||
"request_type": t.request_type,
|
||||
"language": t.language,
|
||||
"latest_version": {
|
||||
"id": str(latest.id),
|
||||
"version": latest.version,
|
||||
"subject": latest.subject,
|
||||
"body_html": latest.body_html,
|
||||
"body_text": latest.body_text,
|
||||
} if latest else None,
|
||||
})
|
||||
return result
|
||||
|
||||
def get_template_versions(self, template_id: str, tenant_id: str) -> List[Dict[str, Any]]:
|
||||
try:
|
||||
tid = uuid.UUID(template_id)
|
||||
except ValueError:
|
||||
raise ValidationError("Invalid template ID")
|
||||
template = self._db.query(DSRTemplateDB).filter(
|
||||
DSRTemplateDB.id == tid,
|
||||
DSRTemplateDB.tenant_id == uuid.UUID(tenant_id),
|
||||
).first()
|
||||
if not template:
|
||||
raise NotFoundError("Template not found")
|
||||
versions = self._db.query(DSRTemplateVersionDB).filter(
|
||||
DSRTemplateVersionDB.template_id == tid,
|
||||
).order_by(DSRTemplateVersionDB.created_at.desc()).all()
|
||||
return [
|
||||
{
|
||||
"id": str(v.id),
|
||||
"template_id": str(v.template_id),
|
||||
"version": v.version,
|
||||
"subject": v.subject,
|
||||
"body_html": v.body_html,
|
||||
"body_text": v.body_text,
|
||||
"status": v.status,
|
||||
"published_at": v.published_at.isoformat() if v.published_at else None,
|
||||
"published_by": v.published_by,
|
||||
"created_at": v.created_at.isoformat() if v.created_at else None,
|
||||
"created_by": v.created_by,
|
||||
}
|
||||
for v in versions
|
||||
]
|
||||
|
||||
def create_template_version(
|
||||
self, template_id: str, body: CreateTemplateVersion, tenant_id: str,
|
||||
) -> Dict[str, Any]:
|
||||
try:
|
||||
tid = uuid.UUID(template_id)
|
||||
except ValueError:
|
||||
raise ValidationError("Invalid template ID")
|
||||
template = self._db.query(DSRTemplateDB).filter(
|
||||
DSRTemplateDB.id == tid,
|
||||
DSRTemplateDB.tenant_id == uuid.UUID(tenant_id),
|
||||
).first()
|
||||
if not template:
|
||||
raise NotFoundError("Template not found")
|
||||
version = DSRTemplateVersionDB(
|
||||
template_id=tid,
|
||||
version=body.version,
|
||||
subject=body.subject,
|
||||
body_html=body.body_html,
|
||||
body_text=body.body_text,
|
||||
status="draft",
|
||||
)
|
||||
self._db.add(version)
|
||||
self._db.commit()
|
||||
self._db.refresh(version)
|
||||
return {
|
||||
"id": str(version.id),
|
||||
"template_id": str(version.template_id),
|
||||
"version": version.version,
|
||||
"subject": version.subject,
|
||||
"body_html": version.body_html,
|
||||
"body_text": version.body_text,
|
||||
"status": version.status,
|
||||
"created_at": version.created_at.isoformat() if version.created_at else None,
|
||||
}
|
||||
|
||||
def publish_template_version(self, version_id: str, tenant_id: str) -> Dict[str, Any]:
|
||||
try:
|
||||
vid = uuid.UUID(version_id)
|
||||
except ValueError:
|
||||
raise ValidationError("Invalid version ID")
|
||||
version = self._db.query(DSRTemplateVersionDB).filter(
|
||||
DSRTemplateVersionDB.id == vid,
|
||||
).first()
|
||||
if not version:
|
||||
raise NotFoundError("Version not found")
|
||||
now = datetime.now(timezone.utc)
|
||||
version.status = "published"
|
||||
version.published_at = now
|
||||
version.published_by = "admin"
|
||||
self._db.commit()
|
||||
self._db.refresh(version)
|
||||
return {
|
||||
"id": str(version.id),
|
||||
"template_id": str(version.template_id),
|
||||
"version": version.version,
|
||||
"status": version.status,
|
||||
"published_at": version.published_at.isoformat(),
|
||||
"published_by": version.published_by,
|
||||
}
|
||||
Reference in New Issue
Block a user