Files
breakpilot-compliance/backend-compliance/compliance/schemas/canonical_control.py
Sharang Parnerkar b850368ec9 refactor(backend/api): extract CanonicalControlService (Step 4 — file 6 of 18)
compliance/api/canonical_control_routes.py (514 LOC) -> 192 LOC thin
routes + 316-line CanonicalControlService + 105-line schemas file.

Canonical Control Library manages OWASP/NIST/ENISA-anchored security
control frameworks and controls. Like company_profile_routes, this file
uses raw SQL via sqlalchemy.text() because there are no SQLAlchemy
models for canonical_control_frameworks or canonical_controls.

Single-service split. Session management moved from bespoke
`with SessionLocal() as db:` blocks to Depends(get_db) for consistency.

Legacy test imports preserved via re-export (FrameworkResponse,
ControlResponse, SimilarityCheckRequest, SimilarityCheckResponse,
_control_row).

Validation extracted to a module-level `_validate_control_input` helper
so both create and update share the same checks. ValidationError (from
compliance.domain) replaces raw HTTPException(400) raises.

Verified:
  - 187/187 pytest (173 core + 14 canonical) pass
  - OpenAPI 360/484 unchanged
  - mypy compliance/ -> Success on 130 source files
  - canonical_control_routes.py 514 -> 192 LOC
  - Hard-cap violations: 13 -> 12

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

106 lines
2.6 KiB
Python

"""
Canonical Control Library schemas.
Phase 1 Step 4: extracted from ``compliance.api.canonical_control_routes``.
"""
from typing import Any, Optional
from pydantic import BaseModel
class FrameworkResponse(BaseModel):
id: str
framework_id: str
name: str
version: str
description: Optional[str] = None
owner: Optional[str] = None
policy_version: Optional[str] = None
release_state: str
created_at: str
updated_at: str
class ControlResponse(BaseModel):
id: str
framework_id: str
control_id: str
title: str
objective: str
rationale: str
scope: dict[str, Any]
requirements: list[Any]
test_procedure: list[Any]
evidence: list[Any]
severity: str
risk_score: Optional[float] = None
implementation_effort: Optional[str] = None
evidence_confidence: Optional[float] = None
open_anchors: list[Any]
release_state: str
tags: list[Any]
created_at: str
updated_at: str
class ControlCreateRequest(BaseModel):
framework_id: str # e.g. 'bp_security_v1'
control_id: str # e.g. 'AUTH-003'
title: str
objective: str
rationale: str
scope: dict[str, Any] = {}
requirements: list[Any] = []
test_procedure: list[Any] = []
evidence: list[Any] = []
severity: str = "medium"
risk_score: Optional[float] = None
implementation_effort: Optional[str] = None
evidence_confidence: Optional[float] = None
open_anchors: list[Any] = []
release_state: str = "draft"
tags: list[Any] = []
class ControlUpdateRequest(BaseModel):
title: Optional[str] = None
objective: Optional[str] = None
rationale: Optional[str] = None
scope: Optional[dict[str, Any]] = None
requirements: Optional[list[Any]] = None
test_procedure: Optional[list[Any]] = None
evidence: Optional[list[Any]] = None
severity: Optional[str] = None
risk_score: Optional[float] = None
implementation_effort: Optional[str] = None
evidence_confidence: Optional[float] = None
open_anchors: Optional[list[Any]] = None
release_state: Optional[str] = None
tags: Optional[list[Any]] = None
class SimilarityCheckRequest(BaseModel):
source_text: str
candidate_text: str
class SimilarityCheckResponse(BaseModel):
max_exact_run: int
token_overlap: float
ngram_jaccard: float
embedding_cosine: float
lcs_ratio: float
status: str
details: dict[str, Any]
__all__ = [
"FrameworkResponse",
"ControlResponse",
"ControlCreateRequest",
"ControlUpdateRequest",
"SimilarityCheckRequest",
"SimilarityCheckResponse",
]