refactor: remove dead code, hollow stubs, and orphaned modules (#2)
All checks were successful
Build + Deploy / build-admin-compliance (push) Successful in 1m40s
Build + Deploy / build-backend-compliance (push) Successful in 2m52s
Build + Deploy / build-ai-sdk (push) Successful in 40s
Build + Deploy / build-developer-portal (push) Successful in 1m2s
Build + Deploy / build-tts (push) Successful in 1m23s
Build + Deploy / build-document-crawler (push) Successful in 40s
Build + Deploy / build-dsms-gateway (push) Successful in 25s
CI / branch-name (push) Has been skipped
CI / guardrail-integrity (push) Has been skipped
CI / loc-budget (push) Successful in 17s
CI / secret-scan (push) Has been skipped
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / nodejs-build (push) Successful in 2m12s
CI / dep-audit (push) Has been skipped
CI / sbom-scan (push) Has been skipped
CI / test-go (push) Successful in 44s
CI / test-python-backend (push) Successful in 37s
CI / test-python-document-crawler (push) Successful in 27s
CI / test-python-dsms-gateway (push) Successful in 29s
CI / validate-canonical-controls (push) Successful in 16s
Build + Deploy / trigger-orca (push) Successful in 2m46s

This commit was merged in pull request #2.
This commit is contained in:
2026-04-20 05:50:59 +00:00
parent f96536ebbe
commit 5231490ccc
13 changed files with 11 additions and 4063 deletions

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@@ -1,63 +0,0 @@
"""
AI System (AI Act) 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,
)
# ============================================================================
# AI System Schemas (AI Act Compliance)
# ============================================================================
class AISystemBase(BaseModel):
name: str
description: Optional[str] = None
purpose: Optional[str] = None
sector: Optional[str] = None
classification: str = "unclassified"
status: str = "draft"
obligations: Optional[List[str]] = None
class AISystemCreate(AISystemBase):
pass
class AISystemUpdate(BaseModel):
name: Optional[str] = None
description: Optional[str] = None
purpose: Optional[str] = None
sector: Optional[str] = None
classification: Optional[str] = None
status: Optional[str] = None
obligations: Optional[List[str]] = None
class AISystemResponse(AISystemBase):
id: str
assessment_date: Optional[datetime] = None
assessment_result: Optional[Dict[str, Any]] = None
risk_factors: Optional[List[Dict[str, Any]]] = None
recommendations: Optional[List[str]] = None
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class AISystemListResponse(BaseModel):
systems: List[AISystemResponse]
total: int

View File

@@ -1,58 +0,0 @@
"""
BSI / PDF Extraction 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,
)
# ============================================================================
# PDF Extraction Schemas
# ============================================================================
class BSIAspectResponse(BaseModel):
"""A single extracted BSI-TR Pruefaspekt (test aspect)."""
aspect_id: str
title: str
full_text: str
category: str
page_number: int
section: str
requirement_level: str
source_document: str
keywords: Optional[List[str]] = None
related_aspects: Optional[List[str]] = None
class PDFExtractionRequest(BaseModel):
"""Request for PDF extraction."""
document_code: str = Field(..., description="BSI-TR document code, e.g. BSI-TR-03161-2")
save_to_db: bool = Field(True, description="Whether to save extracted requirements to database")
force: bool = Field(False, description="Force re-extraction even if requirements exist")
class PDFExtractionResponse(BaseModel):
"""Response from PDF extraction endpoint."""
# Simple endpoint format (new /pdf/extract/{doc_code})
doc_code: Optional[str] = None
total_extracted: Optional[int] = None
saved_to_db: Optional[int] = None
aspects: Optional[List[BSIAspectResponse]] = None
# Legacy scraper endpoint format (/scraper/extract-pdf)
success: Optional[bool] = None
source_document: Optional[str] = None
total_aspects: Optional[int] = None
statistics: Optional[Dict[str, Any]] = None
requirements_created: Optional[int] = None

View File

@@ -1,195 +0,0 @@
"""
Dashboard, Export, Executive Dashboard 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,
)
from compliance.schemas.evidence import EvidenceResponse
from compliance.schemas.risk import RiskResponse
# ============================================================================
# Dashboard & Export Schemas
# ============================================================================
class DashboardResponse(BaseModel):
compliance_score: float
total_regulations: int
total_requirements: int
total_controls: int
controls_by_status: Dict[str, int]
controls_by_domain: Dict[str, Dict[str, int]]
total_evidence: int
evidence_by_status: Dict[str, int]
total_risks: int
risks_by_level: Dict[str, int]
recent_activity: List[Dict[str, Any]]
class ExportRequest(BaseModel):
export_type: str = "full" # "full", "controls_only", "evidence_only"
included_regulations: Optional[List[str]] = None
included_domains: Optional[List[str]] = None
date_range_start: Optional[date] = None
date_range_end: Optional[date] = None
class ExportResponse(BaseModel):
id: str
export_type: str
export_name: Optional[str] = None
status: str
requested_by: str
requested_at: datetime
completed_at: Optional[datetime] = None
file_path: Optional[str] = None
file_hash: Optional[str] = None
file_size_bytes: Optional[int] = None
total_controls: Optional[int] = None
total_evidence: Optional[int] = None
compliance_score: Optional[float] = None
error_message: Optional[str] = None
model_config = ConfigDict(from_attributes=True)
class ExportListResponse(BaseModel):
exports: List[ExportResponse]
total: int
# ============================================================================
# Seeding Schemas
# ============================================================================
class SeedRequest(BaseModel):
force: bool = False
class SeedResponse(BaseModel):
success: bool
message: str
counts: Dict[str, int]
class PaginatedEvidenceResponse(BaseModel):
"""Paginated response for evidence."""
data: List[EvidenceResponse]
pagination: PaginationMeta
class PaginatedRiskResponse(BaseModel):
"""Paginated response for risks."""
data: List[RiskResponse]
pagination: PaginationMeta
# ============================================================================
# Executive Dashboard Schemas (Phase 3 - Sprint 1)
# ============================================================================
class TrendDataPoint(BaseModel):
"""A single data point for trend charts."""
date: str # ISO date string
score: float
label: Optional[str] = None # Formatted date for display (e.g., "Jan 26")
class RiskSummary(BaseModel):
"""Summary of a risk for executive display."""
id: str
risk_id: str
title: str
risk_level: str # "low", "medium", "high", "critical"
owner: Optional[str] = None
status: str
category: str
impact: int
likelihood: int
class DeadlineItem(BaseModel):
"""An upcoming deadline for executive display."""
id: str
title: str
deadline: str # ISO date string
days_remaining: int
type: str # "control_review", "evidence_expiry", "audit"
status: str # "on_track", "at_risk", "overdue"
owner: Optional[str] = None
class TeamWorkloadItem(BaseModel):
"""Workload distribution for a team or person."""
name: str
pending_tasks: int
in_progress_tasks: int
completed_tasks: int
total_tasks: int
completion_rate: float
class ExecutiveDashboardResponse(BaseModel):
"""
Executive Dashboard Response
Provides a high-level overview for managers and executives:
- Traffic light status (green/yellow/red)
- Overall compliance score
- 12-month trend data
- Top 5 risks
- Upcoming deadlines
- Team workload distribution
"""
traffic_light_status: str # "green", "yellow", "red"
overall_score: float
score_trend: List[TrendDataPoint]
previous_score: Optional[float] = None
score_change: Optional[float] = None # Positive = improvement
# Counts
total_regulations: int
total_requirements: int
total_controls: int
open_risks: int
# Top items
top_risks: List[RiskSummary]
upcoming_deadlines: List[DeadlineItem]
# Workload
team_workload: List[TeamWorkloadItem]
# Last updated
last_updated: str
class ComplianceSnapshotCreate(BaseModel):
"""Request to create a compliance snapshot."""
notes: Optional[str] = None
class ComplianceSnapshotResponse(BaseModel):
"""Response for a compliance snapshot."""
id: str
snapshot_date: str
overall_score: float
scores_by_regulation: Dict[str, float]
scores_by_domain: Dict[str, float]
total_controls: int
passed_controls: int
failed_controls: int
notes: Optional[str] = None
created_at: str

View File

@@ -1,343 +0,0 @@
"""
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) ---

View File

@@ -1,53 +0,0 @@
"""
Report generation 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,
)
# ============================================================================
# Report Generation Schemas (Phase 3 - Sprint 3)
# ============================================================================
class GenerateReportRequest(BaseModel):
"""Request to generate an audit report."""
session_id: str
report_type: str = "full" # "full", "summary", "non_compliant_only"
include_evidence: bool = True
include_signatures: bool = True
language: str = "de" # "de" or "en"
class ReportGenerationResponse(BaseModel):
"""Response for report generation."""
report_id: str
session_id: str
status: str # "pending", "generating", "completed", "failed"
report_type: str
file_path: Optional[str] = None
file_size_bytes: Optional[int] = None
created_at: datetime
completed_at: Optional[datetime] = None
error_message: Optional[str] = None
class ReportDownloadResponse(BaseModel):
"""Response for report download."""
report_id: str
filename: str
mime_type: str
file_size_bytes: int
download_url: str

View File

@@ -1,121 +0,0 @@
"""
Service Module 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,
)
# ============================================================================
# Service Module Schemas (Sprint 3)
# ============================================================================
class ServiceModuleBase(BaseModel):
"""Base schema for service modules."""
name: str
display_name: str
description: Optional[str] = None
service_type: str
port: Optional[int] = None
technology_stack: Optional[List[str]] = None
repository_path: Optional[str] = None
docker_image: Optional[str] = None
data_categories: Optional[List[str]] = None
processes_pii: bool = False
processes_health_data: bool = False
ai_components: bool = False
criticality: str = "medium"
owner_team: Optional[str] = None
owner_contact: Optional[str] = None
class ServiceModuleCreate(ServiceModuleBase):
"""Schema for creating a service module."""
pass
class ServiceModuleResponse(ServiceModuleBase):
"""Response schema for service module."""
id: str
is_active: bool
compliance_score: Optional[float] = None
last_compliance_check: Optional[datetime] = None
created_at: datetime
updated_at: datetime
regulation_count: Optional[int] = None
risk_count: Optional[int] = None
model_config = ConfigDict(from_attributes=True)
class ServiceModuleListResponse(BaseModel):
"""List response for service modules."""
modules: List[ServiceModuleResponse]
total: int
class ServiceModuleDetailResponse(ServiceModuleResponse):
"""Detailed response including regulations and risks."""
regulations: Optional[List[Dict[str, Any]]] = None
risks: Optional[List[Dict[str, Any]]] = None
class ModuleRegulationMappingBase(BaseModel):
"""Base schema for module-regulation mapping."""
module_id: str
regulation_id: str
relevance_level: str = "medium"
notes: Optional[str] = None
applicable_articles: Optional[List[str]] = None
class ModuleRegulationMappingCreate(ModuleRegulationMappingBase):
"""Schema for creating a module-regulation mapping."""
pass
class ModuleRegulationMappingResponse(ModuleRegulationMappingBase):
"""Response schema for module-regulation mapping."""
id: str
module_name: Optional[str] = None
regulation_code: Optional[str] = None
regulation_name: Optional[str] = None
created_at: datetime
model_config = ConfigDict(from_attributes=True)
class ModuleSeedRequest(BaseModel):
"""Request to seed service modules."""
force: bool = False
class ModuleSeedResponse(BaseModel):
"""Response from seeding service modules."""
success: bool
message: str
modules_created: int
mappings_created: int
class ModuleComplianceOverview(BaseModel):
"""Overview of compliance status for all modules."""
total_modules: int
modules_by_type: Dict[str, int]
modules_by_criticality: Dict[str, int]
modules_processing_pii: int
modules_with_ai: int
average_compliance_score: Optional[float] = None
regulations_coverage: Dict[str, int] # regulation_code -> module_count