Files
breakpilot-compliance/backend-compliance/compliance/schemas/tom.py
Sharang Parnerkar d571412657 refactor(backend/api): extract TOMService (Step 4 — file 3 of 18)
compliance/api/tom_routes.py (609 LOC) -> 215 LOC thin routes +
434-line TOMService. Request bodies (TOMStateBody, TOMMeasureCreate,
TOMMeasureUpdate, TOMMeasureBulkItem, TOMMeasureBulkBody) moved to
compliance/schemas/tom.py (joining the existing response models from
the Step 3 split).

Single-service split (not two like banner): state, measures CRUD + bulk
upsert, stats, export, and version lookups are all tightly coupled
around the TOMMeasureDB aggregate, so splitting would create artificial
boundaries. TOMService is 434 LOC — comfortably under the 500 hard cap.

Domain error mapping:
  - ConflictError   -> 409 (version conflict on state save; duplicate control_id on create)
  - NotFoundError   -> 404 (missing measure on update; missing version)
  - ValidationError -> 400 (missing tenant_id on DELETE /state)

Legacy test compat: the existing tests/test_tom_routes.py imports
TOMMeasureBulkItem, _parse_dt, _measure_to_dict, and DEFAULT_TENANT_ID
directly from compliance.api.tom_routes. All re-exported via __all__ so
the 44-test file runs unchanged.

mypy.ini flips compliance.api.tom_routes from ignore_errors=True to
False. TOMService carries the scoped Column[T] header.

Verified:
  - 217/217 pytest (173 baseline + 44 TOM) pass
  - OpenAPI 360/484 unchanged
  - mypy compliance/ -> Success on 124 source files
  - tom_routes.py 609 -> 215 LOC
  - Hard-cap violations: 16 -> 15

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-07 19:42:17 +02:00

164 lines
5.5 KiB
Python

"""
TOM (Technisch-Organisatorische Maßnahmen) Pydantic schemas — extracted from compliance/api/schemas.py.
Phase 1 Step 3: the monolithic ``compliance.api.schemas`` module is being
split per domain under ``compliance.schemas``. This module is re-exported
from ``compliance.api.schemas`` for backwards compatibility.
"""
from datetime import datetime, date
from typing import Optional, List, Any, Dict
from pydantic import BaseModel, ConfigDict, Field
from compliance.schemas.common import (
PaginationMeta, RegulationType, ControlType, ControlDomain,
ControlStatus, RiskLevel, EvidenceStatus,
)
# ============================================================================
# TOM — Technisch-Organisatorische Massnahmen (Art. 32 DSGVO)
# ============================================================================
# ---- Request bodies (extracted from compliance/api/tom_routes.py) -----------
class TOMStateBody(BaseModel):
"""Request body for POST /tom/state (save with optimistic locking)."""
tenant_id: Optional[str] = None
tenantId: Optional[str] = None # Accept camelCase from frontend
state: Dict[str, Any]
version: Optional[int] = None
def get_tenant_id(self) -> str:
return self.tenant_id or self.tenantId or "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
class TOMMeasureCreate(BaseModel):
"""Request body for POST /tom/measures."""
control_id: str
name: str
description: Optional[str] = None
category: str
type: str
applicability: str = "REQUIRED"
applicability_reason: Optional[str] = None
implementation_status: str = "NOT_IMPLEMENTED"
responsible_person: Optional[str] = None
responsible_department: Optional[str] = None
implementation_date: Optional[str] = None
review_date: Optional[str] = None
review_frequency: Optional[str] = None
priority: Optional[str] = None
complexity: Optional[str] = None
linked_evidence: Optional[List[Any]] = None
evidence_gaps: Optional[List[Any]] = None
related_controls: Optional[Dict[str, Any]] = None
verified_at: Optional[str] = None
verified_by: Optional[str] = None
effectiveness_rating: Optional[str] = None
class TOMMeasureUpdate(BaseModel):
"""Request body for PUT /tom/measures/{id} (all fields optional)."""
name: Optional[str] = None
description: Optional[str] = None
category: Optional[str] = None
type: Optional[str] = None
applicability: Optional[str] = None
applicability_reason: Optional[str] = None
implementation_status: Optional[str] = None
responsible_person: Optional[str] = None
responsible_department: Optional[str] = None
implementation_date: Optional[str] = None
review_date: Optional[str] = None
review_frequency: Optional[str] = None
priority: Optional[str] = None
complexity: Optional[str] = None
linked_evidence: Optional[List[Any]] = None
evidence_gaps: Optional[List[Any]] = None
related_controls: Optional[Dict[str, Any]] = None
verified_at: Optional[str] = None
verified_by: Optional[str] = None
effectiveness_rating: Optional[str] = None
class TOMMeasureBulkItem(BaseModel):
"""Single item in a TOMMeasureBulkBody — no verification fields."""
control_id: str
name: str
description: Optional[str] = None
category: str
type: str
applicability: str = "REQUIRED"
applicability_reason: Optional[str] = None
implementation_status: str = "NOT_IMPLEMENTED"
responsible_person: Optional[str] = None
responsible_department: Optional[str] = None
implementation_date: Optional[str] = None
review_date: Optional[str] = None
review_frequency: Optional[str] = None
priority: Optional[str] = None
complexity: Optional[str] = None
linked_evidence: Optional[List[Any]] = None
evidence_gaps: Optional[List[Any]] = None
related_controls: Optional[Dict[str, Any]] = None
class TOMMeasureBulkBody(BaseModel):
"""Request body for POST /tom/measures/bulk."""
tenant_id: Optional[str] = None
measures: List["TOMMeasureBulkItem"] = []
# ---- Response models --------------------------------------------------------
class TOMStateResponse(BaseModel):
tenant_id: str
state: Dict[str, Any] = {}
version: int = 0
last_modified: Optional[datetime] = None
is_new: bool = False
class TOMMeasureResponse(BaseModel):
id: str
tenant_id: str
control_id: str
name: str
description: Optional[str] = None
category: str
type: str
applicability: str = "REQUIRED"
applicability_reason: Optional[str] = None
implementation_status: str = "NOT_IMPLEMENTED"
responsible_person: Optional[str] = None
responsible_department: Optional[str] = None
implementation_date: Optional[datetime] = None
review_date: Optional[datetime] = None
review_frequency: Optional[str] = None
priority: Optional[str] = None
complexity: Optional[str] = None
linked_evidence: List[Any] = []
evidence_gaps: List[Any] = []
related_controls: Dict[str, Any] = {}
verified_at: Optional[datetime] = None
verified_by: Optional[str] = None
effectiveness_rating: Optional[str] = None
created_by: Optional[str] = None
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
model_config = ConfigDict(from_attributes=True)
class TOMStatsResponse(BaseModel):
total: int = 0
by_status: Dict[str, int] = {}
by_category: Dict[str, int] = {}
overdue_review_count: int = 0
implemented: int = 0
partial: int = 0
not_implemented: int = 0