# mypy: disable-error-code="arg-type,assignment,union-attr,no-any-return,attr-defined,index,call-overload,type-arg,var-annotated,misc,call-arg,return-value" """ Notfallplan workflow service -- incidents and templates. Phase 1 Step 4: extracted from ``compliance.api.notfallplan_routes``. Core CRUD for contacts/scenarios/checklists/exercises/stats lives in ``compliance.services.notfallplan_service``. """ import json import logging from datetime import datetime, timezone from typing import Any, Dict, List, Optional from sqlalchemy import text from sqlalchemy.orm import Session from compliance.domain import NotFoundError, ValidationError from compliance.schemas.notfallplan import ( IncidentCreate, IncidentUpdate, TemplateCreate, TemplateUpdate, ) logger = logging.getLogger(__name__) # ============================================================================ # Row serializers # ============================================================================ def _incident_row(r: Any) -> Dict[str, Any]: return { "id": str(r.id), "tenant_id": r.tenant_id, "title": r.title, "description": r.description, "detected_at": r.detected_at.isoformat() if r.detected_at else None, "detected_by": r.detected_by, "status": r.status, "severity": r.severity, "affected_data_categories": ( r.affected_data_categories if r.affected_data_categories else [] ), "estimated_affected_persons": r.estimated_affected_persons, "measures": r.measures if r.measures else [], "art34_required": r.art34_required, "art34_justification": r.art34_justification, "reported_to_authority_at": ( r.reported_to_authority_at.isoformat() if r.reported_to_authority_at else None ), "notified_affected_at": ( r.notified_affected_at.isoformat() if r.notified_affected_at else None ), "closed_at": r.closed_at.isoformat() if r.closed_at else None, "closed_by": r.closed_by, "lessons_learned": r.lessons_learned, "created_at": r.created_at.isoformat() if r.created_at else None, "updated_at": r.updated_at.isoformat() if r.updated_at else None, } def _template_row(r: Any) -> Dict[str, Any]: return { "id": str(r.id), "tenant_id": r.tenant_id, "type": r.type, "title": r.title, "content": r.content, "created_at": r.created_at.isoformat() if r.created_at else None, "updated_at": r.updated_at.isoformat() if r.updated_at else None, } class NotfallplanWorkflowService: """Incident and template operations.""" def __init__(self, db: Session) -> None: self.db = db # --------------------------------------------------------------- incidents def list_incidents( self, tenant_id: str, status: Optional[str] = None, severity: Optional[str] = None, ) -> List[Dict[str, Any]]: where = "WHERE tenant_id = :tenant_id" params: Dict[str, Any] = {"tenant_id": tenant_id} if status: where += " AND status = :status" params["status"] = status if severity: where += " AND severity = :severity" params["severity"] = severity rows = self.db.execute( text(f""" SELECT * FROM compliance_notfallplan_incidents {where} ORDER BY created_at DESC """), params, ).fetchall() return [_incident_row(r) for r in rows] def create_incident( self, tenant_id: str, req: IncidentCreate, ) -> Dict[str, Any]: row = self.db.execute( text(""" INSERT INTO compliance_notfallplan_incidents (tenant_id, title, description, detected_by, status, severity, affected_data_categories, estimated_affected_persons, measures, art34_required, art34_justification) VALUES (:tenant_id, :title, :description, :detected_by, :status, :severity, CAST(:affected_data_categories AS jsonb), :estimated_affected_persons, CAST(:measures AS jsonb), :art34_required, :art34_justification) RETURNING * """), { "tenant_id": tenant_id, "title": req.title, "description": req.description, "detected_by": req.detected_by, "status": req.status, "severity": req.severity, "affected_data_categories": json.dumps( req.affected_data_categories ), "estimated_affected_persons": req.estimated_affected_persons, "measures": json.dumps(req.measures), "art34_required": req.art34_required, "art34_justification": req.art34_justification, }, ).fetchone() self.db.commit() return _incident_row(row) def update_incident( self, tenant_id: str, incident_id: str, req: IncidentUpdate, ) -> Dict[str, Any]: existing = self.db.execute( text( "SELECT id FROM compliance_notfallplan_incidents" " WHERE id = :id AND tenant_id = :tenant_id" ), {"id": incident_id, "tenant_id": tenant_id}, ).fetchone() if not existing: raise NotFoundError(f"Incident {incident_id} not found") updates = req.dict(exclude_none=True) if not updates: raise ValidationError("No fields to update") # Auto-set timestamps based on status transitions if ( updates.get("status") == "reported" and not updates.get("reported_to_authority_at") ): updates["reported_to_authority_at"] = ( datetime.now(timezone.utc).isoformat() ) if ( updates.get("status") == "closed" and not updates.get("closed_at") ): updates["closed_at"] = datetime.now(timezone.utc).isoformat() updates["updated_at"] = datetime.now(timezone.utc).isoformat() set_parts = [] for k in updates: if k in ("affected_data_categories", "measures"): set_parts.append(f"{k} = CAST(:{k} AS jsonb)") updates[k] = ( json.dumps(updates[k]) if isinstance(updates[k], list) else updates[k] ) else: set_parts.append(f"{k} = :{k}") updates["id"] = incident_id updates["tenant_id"] = tenant_id row = self.db.execute( text(f""" UPDATE compliance_notfallplan_incidents SET {', '.join(set_parts)} WHERE id = :id AND tenant_id = :tenant_id RETURNING * """), updates, ).fetchone() self.db.commit() return _incident_row(row) def delete_incident(self, tenant_id: str, incident_id: str) -> None: result = self.db.execute( text( "DELETE FROM compliance_notfallplan_incidents" " WHERE id = :id AND tenant_id = :tenant_id" ), {"id": incident_id, "tenant_id": tenant_id}, ) self.db.commit() if result.rowcount == 0: raise NotFoundError(f"Incident {incident_id} not found") # -------------------------------------------------------------- templates def list_templates( self, tenant_id: str, type: Optional[str] = None, ) -> List[Dict[str, Any]]: where = "WHERE tenant_id = :tenant_id" params: Dict[str, Any] = {"tenant_id": tenant_id} if type: where += " AND type = :type" params["type"] = type rows = self.db.execute( text( f"SELECT * FROM compliance_notfallplan_templates" f" {where} ORDER BY type, created_at" ), params, ).fetchall() return [_template_row(r) for r in rows] def create_template( self, tenant_id: str, req: TemplateCreate, ) -> Dict[str, Any]: row = self.db.execute( text(""" INSERT INTO compliance_notfallplan_templates (tenant_id, type, title, content) VALUES (:tenant_id, :type, :title, :content) RETURNING * """), { "tenant_id": tenant_id, "type": req.type, "title": req.title, "content": req.content, }, ).fetchone() self.db.commit() return _template_row(row) def update_template( self, tenant_id: str, template_id: str, req: TemplateUpdate, ) -> Dict[str, Any]: existing = self.db.execute( text( "SELECT id FROM compliance_notfallplan_templates" " WHERE id = :id AND tenant_id = :tenant_id" ), {"id": template_id, "tenant_id": tenant_id}, ).fetchone() if not existing: raise NotFoundError(f"Template {template_id} not found") updates = req.dict(exclude_none=True) if not updates: raise ValidationError("No fields to update") updates["updated_at"] = datetime.now(timezone.utc).isoformat() set_clauses = ", ".join(f"{k} = :{k}" for k in updates) updates["id"] = template_id updates["tenant_id"] = tenant_id row = self.db.execute( text(f""" UPDATE compliance_notfallplan_templates SET {set_clauses} WHERE id = :id AND tenant_id = :tenant_id RETURNING * """), updates, ).fetchone() self.db.commit() return _template_row(row) def delete_template(self, tenant_id: str, template_id: str) -> None: result = self.db.execute( text( "DELETE FROM compliance_notfallplan_templates" " WHERE id = :id AND tenant_id = :tenant_id" ), {"id": template_id, "tenant_id": tenant_id}, ) self.db.commit() if result.rowcount == 0: raise NotFoundError(f"Template {template_id} not found")