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

refactor(backend/api): split schemas.py into per-domain modules (1899 -> 39 LOC shim)

Browse Source 
Phase 1 Step 3 of PHASE1_RUNBOOK.md. compliance/api/schemas.py is
decomposed into 16 per-domain Pydantic schema modules under
compliance/schemas/:

  common.py          ( 79) — 6 API enums + PaginationMeta
  regulation.py      ( 52)
  requirement.py     ( 80)
  control.py         (119) — Control + Mapping
  evidence.py        ( 66)
  risk.py            ( 79)
  ai_system.py       ( 63)
  dashboard.py       (195) — Dashboard, Export, Executive Dashboard
  service_module.py  (121)
  bsi.py             ( 58) — BSI + PDF extraction
  audit_session.py   (172)
  report.py          ( 53)
  isms_governance.py (343) — Scope, Context, Policy, Objective, SoA
  isms_audit.py      (431) — Finding, CAPA, Review, Internal Audit, Readiness, Trail, ISO27001
  vvt.py             (168)
  tom.py             ( 71)

compliance/api/schemas.py becomes a 39-line re-export shim so existing
imports (from compliance.api.schemas import RegulationResponse) keep
working unchanged. New code should import from the domain module
directly (from compliance.schemas.regulation import RegulationResponse).

Deferred-from-sweep: all 28 class Config blocks in the original file
were converted to model_config = ConfigDict(...) during the split.
schemas.py-sourced PydanticDeprecatedSince20 warnings are now gone.

Cross-domain references handled via targeted imports (e.g. dashboard.py
imports EvidenceResponse from evidence, RiskResponse from risk). common
API enums + PaginationMeta are imported by every domain module.

Verified:
  - 173/173 pytest compliance/tests/ tests/contracts/ pass
  - OpenAPI 360 paths / 484 operations unchanged (contract test green)
  - All new files under the 500-line hard cap (largest: isms_audit.py
    at 431, isms_governance.py at 343, dashboard.py at 195)
  - No file in compliance/schemas/ or compliance/api/schemas.py
    exceeds the hard cap

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sharang Parnerkar
2026-04-07 18:06:27 +02:00
parent 3320ef94fc
commit d9dcfb97ef
17 changed files with 2186 additions and 1896 deletions
Download Patch File Download Diff File Expand all files Collapse all files

343
backend-compliance/compliance/schemas/isms_governance.py Normal file
View File

@@ -0,0 +1,343 @@
"""
ISMS Governance (Scope, Context, Policy, Objective, SoA) 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,
)
# ============================================================================
# ISO 27001 ISMS Schemas (Kapitel 4-10)
# ============================================================================
# --- Enums ---
class ApprovalStatus(str):
DRAFT = "draft"
UNDER_REVIEW = "under_review"
APPROVED = "approved"
SUPERSEDED = "superseded"
class FindingType(str):
MAJOR = "major"
MINOR = "minor"
OFI = "ofi"
POSITIVE = "positive"
class FindingStatus(str):
OPEN = "open"
IN_PROGRESS = "in_progress"
CAPA_PENDING = "capa_pending"
VERIFICATION_PENDING = "verification_pending"
VERIFIED = "verified"
CLOSED = "closed"
class CAPAType(str):
CORRECTIVE = "corrective"
PREVENTIVE = "preventive"
BOTH = "both"
# --- ISMS Scope (ISO 27001 4.3) ---
class ISMSScopeBase(BaseModel):
"""Base schema for ISMS Scope."""
scope_statement: str
included_locations: Optional[List[str]] = None
included_processes: Optional[List[str]] = None
included_services: Optional[List[str]] = None
excluded_items: Optional[List[str]] = None
exclusion_justification: Optional[str] = None
organizational_boundary: Optional[str] = None
physical_boundary: Optional[str] = None
technical_boundary: Optional[str] = None
class ISMSScopeCreate(ISMSScopeBase):
"""Schema for creating ISMS Scope."""
pass
class ISMSScopeUpdate(BaseModel):
"""Schema for updating ISMS Scope."""
scope_statement: Optional[str] = None
included_locations: Optional[List[str]] = None
included_processes: Optional[List[str]] = None
included_services: Optional[List[str]] = None
excluded_items: Optional[List[str]] = None
exclusion_justification: Optional[str] = None
organizational_boundary: Optional[str] = None
physical_boundary: Optional[str] = None
technical_boundary: Optional[str] = None
class ISMSScopeResponse(ISMSScopeBase):
"""Response schema for ISMS Scope."""
id: str
version: str
status: str
approved_by: Optional[str] = None
approved_at: Optional[datetime] = None
effective_date: Optional[date] = None
review_date: Optional[date] = None
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class ISMSScopeApproveRequest(BaseModel):
"""Request to approve ISMS Scope."""
approved_by: str
effective_date: date
review_date: date
# --- ISMS Context (ISO 27001 4.1, 4.2) ---
class ContextIssue(BaseModel):
"""Single context issue."""
issue: str
impact: str
treatment: Optional[str] = None
class InterestedParty(BaseModel):
"""Single interested party."""
party: str
requirements: List[str]
relevance: str
class ISMSContextBase(BaseModel):
"""Base schema for ISMS Context."""
internal_issues: Optional[List[ContextIssue]] = None
external_issues: Optional[List[ContextIssue]] = None
interested_parties: Optional[List[InterestedParty]] = None
regulatory_requirements: Optional[List[str]] = None
contractual_requirements: Optional[List[str]] = None
swot_strengths: Optional[List[str]] = None
swot_weaknesses: Optional[List[str]] = None
swot_opportunities: Optional[List[str]] = None
swot_threats: Optional[List[str]] = None
class ISMSContextCreate(ISMSContextBase):
"""Schema for creating ISMS Context."""
pass
class ISMSContextResponse(ISMSContextBase):
"""Response schema for ISMS Context."""
id: str
version: str
status: str
approved_by: Optional[str] = None
approved_at: Optional[datetime] = None
last_reviewed_at: Optional[datetime] = None
next_review_date: Optional[date] = None
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
# --- ISMS Policies (ISO 27001 5.2) ---
class ISMSPolicyBase(BaseModel):
"""Base schema for ISMS Policy."""
policy_id: str
title: str
policy_type: str # "master", "operational", "technical"
description: Optional[str] = None
policy_text: str
applies_to: Optional[List[str]] = None
review_frequency_months: int = 12
related_controls: Optional[List[str]] = None
class ISMSPolicyCreate(ISMSPolicyBase):
"""Schema for creating ISMS Policy."""
authored_by: str
class ISMSPolicyUpdate(BaseModel):
"""Schema for updating ISMS Policy."""
title: Optional[str] = None
description: Optional[str] = None
policy_text: Optional[str] = None
applies_to: Optional[List[str]] = None
review_frequency_months: Optional[int] = None
related_controls: Optional[List[str]] = None
class ISMSPolicyResponse(ISMSPolicyBase):
"""Response schema for ISMS Policy."""
id: str
version: str
status: str
authored_by: Optional[str] = None
reviewed_by: Optional[str] = None
approved_by: Optional[str] = None
approved_at: Optional[datetime] = None
effective_date: Optional[date] = None
next_review_date: Optional[date] = None
document_path: Optional[str] = None
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class ISMSPolicyListResponse(BaseModel):
"""List response for ISMS Policies."""
policies: List[ISMSPolicyResponse]
total: int
class ISMSPolicyApproveRequest(BaseModel):
"""Request to approve ISMS Policy."""
reviewed_by: str
approved_by: str
effective_date: date
# --- Security Objectives (ISO 27001 6.2) ---
class SecurityObjectiveBase(BaseModel):
"""Base schema for Security Objective."""
objective_id: str
title: str
description: Optional[str] = None
category: str # "availability", "confidentiality", "integrity", "compliance"
specific: Optional[str] = None
measurable: Optional[str] = None
achievable: Optional[str] = None
relevant: Optional[str] = None
time_bound: Optional[str] = None
kpi_name: Optional[str] = None
kpi_target: Optional[str] = None
kpi_unit: Optional[str] = None
measurement_frequency: Optional[str] = None
owner: Optional[str] = None
target_date: Optional[date] = None
related_controls: Optional[List[str]] = None
related_risks: Optional[List[str]] = None
class SecurityObjectiveCreate(SecurityObjectiveBase):
"""Schema for creating Security Objective."""
pass
class SecurityObjectiveUpdate(BaseModel):
"""Schema for updating Security Objective."""
title: Optional[str] = None
description: Optional[str] = None
kpi_current: Optional[str] = None
progress_percentage: Optional[int] = None
status: Optional[str] = None
class SecurityObjectiveResponse(SecurityObjectiveBase):
"""Response schema for Security Objective."""
id: str
kpi_current: Optional[str] = None
status: str
progress_percentage: int
achieved_date: Optional[date] = None
approved_by: Optional[str] = None
approved_at: Optional[datetime] = None
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class SecurityObjectiveListResponse(BaseModel):
"""List response for Security Objectives."""
objectives: List[SecurityObjectiveResponse]
total: int
# --- Statement of Applicability (SoA) ---
class SoAEntryBase(BaseModel):
"""Base schema for SoA Entry."""
annex_a_control: str # e.g., "A.5.1"
annex_a_title: str
annex_a_category: Optional[str] = None
is_applicable: bool
applicability_justification: str
implementation_status: str = "planned"
implementation_notes: Optional[str] = None
breakpilot_control_ids: Optional[List[str]] = None
coverage_level: str = "full"
evidence_description: Optional[str] = None
risk_assessment_notes: Optional[str] = None
compensating_controls: Optional[str] = None
class SoAEntryCreate(SoAEntryBase):
"""Schema for creating SoA Entry."""
pass
class SoAEntryUpdate(BaseModel):
"""Schema for updating SoA Entry."""
is_applicable: Optional[bool] = None
applicability_justification: Optional[str] = None
implementation_status: Optional[str] = None
implementation_notes: Optional[str] = None
breakpilot_control_ids: Optional[List[str]] = None
coverage_level: Optional[str] = None
evidence_description: Optional[str] = None
class SoAEntryResponse(SoAEntryBase):
"""Response schema for SoA Entry."""
id: str
evidence_ids: Optional[List[str]] = None
reviewed_by: Optional[str] = None
reviewed_at: Optional[datetime] = None
approved_by: Optional[str] = None
approved_at: Optional[datetime] = None
version: str
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class SoAListResponse(BaseModel):
"""List response for SoA."""
entries: List[SoAEntryResponse]
total: int
applicable_count: int
not_applicable_count: int
implemented_count: int
planned_count: int
class SoAApproveRequest(BaseModel):
"""Request to approve SoA entry."""
reviewed_by: str
approved_by: str
# --- Audit Findings (Major/Minor/OFI) ---