feat(template-rules): backend foundation for profile-based document recommendations
CI / detect-changes (push) Successful in 12s
CI / branch-name (push) Has been skipped
CI / test-python-backend (push) Successful in 33s
CI / test-python-document-crawler (push) Successful in 23s
CI / test-python-dsms-gateway (push) Successful in 19s
CI / guardrail-integrity (push) Has been skipped
CI / secret-scan (push) Has been skipped
CI / dep-audit (push) Has been skipped
CI / sbom-scan (push) Has been skipped
CI / build-sha-integrity (push) Failing after 7s
CI / validate-canonical-controls (push) Successful in 16s
CI / loc-budget (push) Failing after 18s
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 2m27s
CI / test-go (push) Failing after 46s
CI / iace-gt-coverage (push) Successful in 25s

Introduces the sustainable backend replacement for the hardcoded inline rules in
admin-compliance/app/sdk/document-generator/templateRecommendations.ts.

What's in this commit (Phase 1.1 - 1.5 of the rustling-yawning-boot plan):

- Migration 147: 4 new tables
  - compliance_template_rules (rule shell, document_type, current_version_id)
  - compliance_template_rule_versions (lifecycle, JSONB conditions,
    source_citation, change_summary, approval timestamps)
  - compliance_template_rule_approvals (audit trail)
  - compliance_tenant_rule_overrides (per-tenant classification overrides)
  Plus partial unique index for "only one is_live=1 version per rule".

- SQLAlchemy models: TemplateRuleDB, TemplateRuleVersionDB,
  TemplateRuleApprovalDB, TenantRuleOverrideDB (compliance/db/).

- Pydantic schemas (compliance/schemas/template_rule.py): full request/response
  set including RecommendationRequest/Result with reasons and override tracking.

- TemplateRuleService (compliance/services/): CRUD + Lifecycle transitions
  (submit_for_review/approve/publish/reject) following legal_document_service.py
  pattern with _transition() helper and approval audit trail. Plus tenant
  override upsert.

- RecommendationService: condition evaluator (eq, neq, in, not_in, gte/lte/gt/lt,
  exists, truthy) over JSONB conditions, override application, reason generation
  for human-readable explanations in workspace UI.

- 18 FastAPI routes in compliance/api/template_rule_routes.py covering rule CRUD,
  version lifecycle, override management and POST /recommend evaluation endpoint.

- Seed data: 33 initial rules ported from templateRecommendations.ts in
  compliance/data/template_rule_seed_data.py, written as published versions
  on first seed run. Idempotent via rule_key.

Phase 1.6 (pytest suite) and Phase 2 (editorial UI in admin-compliance) follow
in separate commits.

[migration-approved]

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-06-06 23:13:50 +02:00
parent 37093ff9e3
commit bb183b0e75
9 changed files with 1839 additions and 0 deletions
@@ -73,6 +73,7 @@ _ROUTER_MODULES = [
"tcf_routes",
"founding_wizard_routes",
"licenses_routes",
"template_rule_routes",
]
_loaded_count = 0
@@ -0,0 +1,263 @@
"""
FastAPI routes for Template Rules — Empfehlungs-Regeln + Versionierung +
Approval-Workflow + Tenant-Overrides + Recommend.
Mounted unter /api/v1/compliance/* (Tenant via X-Tenant-ID Header).
"""
import logging
from typing import Optional
from fastapi import APIRouter, Depends, Header, Query
from sqlalchemy.orm import Session
from classroom_engine.database import get_db
from compliance.api._http_errors import translate_domain_errors
from compliance.schemas.template_rule import (
ApprovalActionRequest,
ApprovalHistoryEntry,
OverrideCreate,
OverrideResponse,
RecommendationRequest,
RecommendationResult,
RejectActionRequest,
RuleCreate,
RuleResponse,
RuleVersionCreate,
RuleVersionResponse,
RuleVersionUpdate,
SubmitForReviewRequest,
)
from compliance.services.recommendation_service import RecommendationService
from compliance.services.template_rule_service import TemplateRuleService
logger = logging.getLogger(__name__)
router = APIRouter(prefix="", tags=["template-rules"])
DEFAULT_TENANT = "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
def _get_tenant(
x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID"),
) -> str:
return x_tenant_id or DEFAULT_TENANT
def _rule_svc(db: Session = Depends(get_db)) -> TemplateRuleService:
return TemplateRuleService(db)
def _rec_svc(db: Session = Depends(get_db)) -> RecommendationService:
return RecommendationService(db)
# ============================================================================
# Rules (Hülle)
# ============================================================================
@router.get("/template-rules", response_model=list[RuleResponse])
async def list_rules(
document_type: Optional[str] = Query(None),
svc: TemplateRuleService = Depends(_rule_svc),
) -> list[RuleResponse]:
with translate_domain_errors():
return svc.list_rules(document_type=document_type)
@router.post("/template-rules", response_model=RuleResponse, status_code=201)
async def create_rule(
request: RuleCreate,
svc: TemplateRuleService = Depends(_rule_svc),
) -> RuleResponse:
with translate_domain_errors():
return svc.create_rule(request)
@router.get("/template-rules/{rule_id}", response_model=RuleResponse)
async def get_rule(
rule_id: str, svc: TemplateRuleService = Depends(_rule_svc),
) -> RuleResponse:
with translate_domain_errors():
return svc.get_rule(rule_id)
@router.delete("/template-rules/{rule_id}", status_code=204)
async def delete_rule(
rule_id: str, svc: TemplateRuleService = Depends(_rule_svc),
) -> None:
with translate_domain_errors():
svc.delete_rule(rule_id)
# ============================================================================
# Versions
# ============================================================================
@router.get(
"/template-rules/{rule_id}/versions",
response_model=list[RuleVersionResponse],
)
async def list_versions(
rule_id: str, svc: TemplateRuleService = Depends(_rule_svc),
) -> list[RuleVersionResponse]:
with translate_domain_errors():
return svc.list_versions_for(rule_id)
@router.post(
"/template-rules/{rule_id}/versions",
response_model=RuleVersionResponse,
status_code=201,
)
async def create_version(
rule_id: str,
request: RuleVersionCreate,
svc: TemplateRuleService = Depends(_rule_svc),
) -> RuleVersionResponse:
request.rule_id = rule_id # URL gewinnt
with translate_domain_errors():
return svc.create_version(request)
@router.get(
"/template-rule-versions/{version_id}",
response_model=RuleVersionResponse,
)
async def get_version(
version_id: str, svc: TemplateRuleService = Depends(_rule_svc),
) -> RuleVersionResponse:
with translate_domain_errors():
return svc.get_version(version_id)
@router.patch(
"/template-rule-versions/{version_id}",
response_model=RuleVersionResponse,
)
async def update_version(
version_id: str,
request: RuleVersionUpdate,
svc: TemplateRuleService = Depends(_rule_svc),
) -> RuleVersionResponse:
with translate_domain_errors():
return svc.update_version(version_id, request)
@router.post(
"/template-rule-versions/{version_id}/submit-review",
response_model=RuleVersionResponse,
)
async def submit_for_review(
version_id: str,
request: SubmitForReviewRequest,
svc: TemplateRuleService = Depends(_rule_svc),
) -> RuleVersionResponse:
with translate_domain_errors():
return svc.submit_for_review(version_id, request)
@router.post(
"/template-rule-versions/{version_id}/approve",
response_model=RuleVersionResponse,
)
async def approve_version(
version_id: str,
request: ApprovalActionRequest,
svc: TemplateRuleService = Depends(_rule_svc),
) -> RuleVersionResponse:
with translate_domain_errors():
return svc.approve(version_id, request)
@router.post(
"/template-rule-versions/{version_id}/publish",
response_model=RuleVersionResponse,
)
async def publish_version(
version_id: str,
request: ApprovalActionRequest,
svc: TemplateRuleService = Depends(_rule_svc),
) -> RuleVersionResponse:
with translate_domain_errors():
return svc.publish(version_id, request)
@router.post(
"/template-rule-versions/{version_id}/reject",
response_model=RuleVersionResponse,
)
async def reject_version(
version_id: str,
request: RejectActionRequest,
svc: TemplateRuleService = Depends(_rule_svc),
) -> RuleVersionResponse:
with translate_domain_errors():
return svc.reject(version_id, request)
@router.get(
"/template-rule-versions/{version_id}/approval-history",
response_model=list[ApprovalHistoryEntry],
)
async def approval_history(
version_id: str, svc: TemplateRuleService = Depends(_rule_svc),
) -> list[ApprovalHistoryEntry]:
with translate_domain_errors():
return svc.approval_history(version_id)
# ============================================================================
# Tenant Overrides
# ============================================================================
@router.get(
"/tenant-rule-overrides",
response_model=list[OverrideResponse],
)
async def list_overrides(
tenant_id: str = Depends(_get_tenant),
svc: TemplateRuleService = Depends(_rule_svc),
) -> list[OverrideResponse]:
with translate_domain_errors():
return svc.list_overrides(tenant_id)
@router.post(
"/tenant-rule-overrides",
response_model=OverrideResponse,
status_code=201,
)
async def upsert_override(
request: OverrideCreate,
tenant_id: str = Depends(_get_tenant),
svc: TemplateRuleService = Depends(_rule_svc),
) -> OverrideResponse:
with translate_domain_errors():
return svc.upsert_override(tenant_id, request)
@router.delete(
"/tenant-rule-overrides/{override_id}",
status_code=204,
)
async def delete_override(
override_id: str,
tenant_id: str = Depends(_get_tenant),
svc: TemplateRuleService = Depends(_rule_svc),
) -> None:
with translate_domain_errors():
svc.delete_override(tenant_id, override_id)
# ============================================================================
# Recommend
# ============================================================================
@router.post("/recommend", response_model=RecommendationResult)
async def recommend(
request: RecommendationRequest,
tenant_id: str = Depends(_get_tenant),
svc: RecommendationService = Depends(_rec_svc),
) -> RecommendationResult:
with translate_domain_errors():
return svc.recommend(request, tenant_id=tenant_id)
@@ -0,0 +1,335 @@
"""
Seed-Daten für ``compliance_template_rules`` — die 33 Initial-Regeln, die aus
``admin-compliance/app/sdk/document-generator/templateRecommendations.ts``
übernommen wurden.
Reine Datenstruktur — die Logik liegt in ``scripts/seed_template_rules.py``.
"""
from typing import Any
def _level_clause(*levels: str) -> dict:
return {"field": "compliance_depth_level", "op": "in", "value": list(levels)}
SEED_RULES: list[dict[str, Any]] = [
{
"rule_key": "employee_dsi_required_with_employees",
"document_type": "employee_dsi",
"title": "Datenschutzinformation für Mitarbeiter",
"classification": "required",
"conditions": {"kind": "any", "clauses": [
{"field": "org_has_employees", "op": "eq", "value": "yes"},
{"field": "org_employee_count", "op": "not_in",
"value": ["none", "0", "1_9"]},
]},
"rationale": "Art. 13 DSGVO Informationspflichten gegenüber Beschäftigten.",
},
{
"rule_key": "applicant_dsi_recommended_with_employees",
"document_type": "applicant_dsi",
"title": "Datenschutzinformation für Bewerber",
"classification": "recommended",
"conditions": {"kind": "any", "clauses": [
{"field": "org_has_employees", "op": "eq", "value": "yes"},
{"field": "org_employee_count", "op": "not_in",
"value": ["none", "0"]},
]},
"rationale": "Empfehlenswert, sobald Bewerbungsdaten verarbeitet werden.",
},
{
"rule_key": "whistleblower_required_50plus",
"document_type": "whistleblower_policy",
"title": "Hinweisgeberschutz-Richtlinie (HinSchG)",
"classification": "required",
"conditions": {"kind": "all", "clauses": [
{"field": "org_employee_count", "op": "in",
"value": ["50_249", "250_999", "1000_plus"]},
]},
"rationale": "§ 12 HinSchG — Pflicht ab 50 Beschäftigten.",
},
{
"rule_key": "ai_usage_required_when_ai_used",
"document_type": "ai_usage_policy",
"title": "KI-Nutzungsrichtlinie",
"classification": "required",
"conditions": {"kind": "any", "clauses": [
{"field": "proc_ai_usage", "op": "not_in", "value": ["none", "no"]},
{"field": "proc_uses_ai_tools", "op": "truthy"},
]},
"rationale": "AI Act + interne Governance bei KI-Einsatz.",
},
{
"rule_key": "byod_required_when_allowed",
"document_type": "byod_policy",
"title": "BYOD-Richtlinie (Bring Your Own Device)",
"classification": "required",
"conditions": {"kind": "all", "clauses": [
{"field": "proc_byod_allowed", "op": "eq", "value": "yes"},
]},
"rationale": "Erforderlich wenn private Geräte für Arbeit genutzt werden.",
},
{
"rule_key": "social_media_dsi_required",
"document_type": "social_media_dsi",
"title": "Datenschutzhinweis Social Media",
"classification": "required",
"conditions": {"kind": "all", "clauses": [
{"field": "org_has_social_media", "op": "eq", "value": "yes"},
]},
"rationale": "BVerfG/EuGH-Rechtsprechung zu gemeinsamer Verantwortlichkeit.",
},
{
"rule_key": "video_conference_dsi_recommended",
"document_type": "video_conference_dsi",
"title": "Datenschutzhinweis Videokonferenz",
"classification": "recommended",
"conditions": {"kind": "all", "clauses": [
{"field": "org_has_video_conferencing", "op": "eq", "value": "yes"},
]},
"rationale": "DSK Orientierungshilfe Videokonferenzsysteme.",
},
{
"rule_key": "information_security_required_l3plus",
"document_type": "information_security_policy",
"title": "Informationssicherheits-Richtlinie",
"classification": "required",
"conditions": {"kind": "all", "clauses": [_level_clause("L3", "L4")]},
"rationale": "Ab Compliance-Tiefe L3 (Strict) erforderlich.",
},
{
"rule_key": "password_recommended_l2plus",
"document_type": "password_policy",
"title": "Passwort-Richtlinie",
"classification": "recommended",
"conditions": {"kind": "all", "clauses": [_level_clause("L2", "L3", "L4")]},
"rationale": "Best Practice IT-Sicherheit ab L2.",
},
{
"rule_key": "encryption_recommended_l3plus",
"document_type": "encryption_policy",
"title": "Verschlüsselungs-Richtlinie",
"classification": "recommended",
"conditions": {"kind": "all", "clauses": [_level_clause("L3", "L4")]},
"rationale": "BSI-Empfehlung, ISO 27001 A.10.",
},
{
"rule_key": "access_control_recommended_l3plus",
"document_type": "access_control_policy",
"title": "Zugriffskontroll-Richtlinie",
"classification": "recommended",
"conditions": {"kind": "all", "clauses": [_level_clause("L3", "L4")]},
"rationale": "Art. 32 DSGVO i.V.m. ISO 27001 A.9.",
},
{
"rule_key": "it_security_required_l3plus",
"document_type": "it_security_concept",
"title": "IT-Sicherheitskonzept",
"classification": "required",
"conditions": {"kind": "all", "clauses": [_level_clause("L3", "L4")]},
"rationale": "BSI IT-Grundschutz / ISO 27001.",
},
{
"rule_key": "backup_recommended_l3plus",
"document_type": "backup_recovery_concept",
"title": "Backup- und Recovery-Konzept",
"classification": "recommended",
"conditions": {"kind": "all", "clauses": [_level_clause("L3", "L4")]},
"rationale": "Art. 32 DSGVO 'Verfügbarkeit + Belastbarkeit'.",
},
{
"rule_key": "logging_recommended_l3plus",
"document_type": "logging_concept",
"title": "Protokollierungskonzept",
"classification": "recommended",
"conditions": {"kind": "all", "clauses": [_level_clause("L3", "L4")]},
"rationale": "Art. 32 DSGVO + BSI IT-Grundschutz.",
},
{
"rule_key": "access_control_concept_recommended_l3plus",
"document_type": "access_control_concept",
"title": "Zugriffskonzept",
"classification": "recommended",
"conditions": {"kind": "all", "clauses": [_level_clause("L3", "L4")]},
"rationale": "ISO 27001 A.9 / BSI IT-Grundschutz ORP.4.",
},
{
"rule_key": "community_guidelines_required_ugc_platform",
"document_type": "community_guidelines",
"title": "Community-Richtlinien",
"classification": "required",
"conditions": {"kind": "any", "clauses": [
{"field": "prod_ugc_platform", "op": "eq", "value": "yes"},
{"field": "org_business_model", "op": "in",
"value": ["platform", "marketplace", "social"]},
]},
"rationale": "DSA + NetzDG für Plattformen mit nutzergeneriertem Inhalt.",
},
{
"rule_key": "terms_of_use_required_platforms",
"document_type": "terms_of_use",
"title": "Nutzungsbedingungen",
"classification": "required",
"conditions": {"kind": "any", "clauses": [
{"field": "prod_ugc_platform", "op": "eq", "value": "yes"},
{"field": "org_business_model", "op": "in",
"value": ["platform", "marketplace", "social", "saas"]},
]},
"rationale": "Plattform-/SaaS-Geschäft braucht klare Nutzungsregeln.",
},
{
"rule_key": "media_content_policy_recommended",
"document_type": "media_content_policy",
"title": "Medien-/Content-Policy",
"classification": "recommended",
"conditions": {"kind": "all", "clauses": [
{"field": "org_business_model", "op": "in", "value": ["platform", "media"]},
]},
"rationale": "Empfehlenswert für Media-/Plattform-Geschäftsmodelle.",
},
{
"rule_key": "widerruf_required_webshop",
"document_type": "widerruf",
"title": "Widerrufsbelehrung",
"classification": "required",
"conditions": {"kind": "all", "clauses": [
{"field": "prod_webshop", "op": "neq", "value": "no"},
]},
"rationale": "§§ 312g, 355 BGB bei Fernabsatzverträgen B2C.",
},
{
"rule_key": "consent_texts_recommended_with_consent_mgmt",
"document_type": "consent_texts",
"title": "Einwilligungstexte (Double-Opt-In)",
"classification": "recommended",
"conditions": {"kind": "all", "clauses": [
{"field": "prod_consent_management", "op": "neq", "value": "no"},
]},
"rationale": "§ 7 UWG + Art. 7 DSGVO.",
},
{
"rule_key": "impressum_always_required",
"document_type": "impressum",
"title": "Impressum",
"classification": "required",
"conditions": {"kind": "all", "clauses": []},
"rationale": "§ 5 TMG / § 18 MStV — gilt für jedes Telemedienangebot.",
},
{
"rule_key": "cookie_policy_always_required",
"document_type": "cookie_policy",
"title": "Cookie-Richtlinie",
"classification": "required",
"conditions": {"kind": "all", "clauses": []},
"rationale": "§ 25 TDDDG + Art. 5 (3) ePrivacy.",
},
{
"rule_key": "privacy_policy_always_required",
"document_type": "privacy_policy",
"title": "Datenschutzerklärung",
"classification": "required",
"conditions": {"kind": "all", "clauses": []},
"rationale": "Art. 13 DSGVO — gilt für jede Verarbeitung.",
},
{
"rule_key": "data_protection_policy_required_l2plus",
"document_type": "data_protection_policy",
"title": "Datenschutzleitlinie",
"classification": "required",
"conditions": {"kind": "all", "clauses": [_level_clause("L2", "L3", "L4")]},
"rationale": "Interne Leitlinie ab Standard-Compliance-Tiefe.",
},
{
"rule_key": "dsfa_required_when_flagged",
"document_type": "dsfa",
"title": "Datenschutz-Folgenabschätzung (DSFA)",
"classification": "required",
"conditions": {"kind": "any", "clauses": [
{"field": "proc_dsfa_required", "op": "eq", "value": "yes"},
{"field": "comp_dsfa_processes", "op": "eq", "value": "required"},
]},
"rationale": "Art. 35 DSGVO + DSK Muss-Liste.",
},
{
"rule_key": "dpa_required_with_processors",
"document_type": "dpa",
"title": "Auftragsverarbeitungsvertrag (AVV)",
"classification": "required",
"conditions": {"kind": "any", "clauses": [
{"field": "comp_has_processors", "op": "neq", "value": "no"},
{"field": "comp_vendor_management", "op": "neq", "value": "no"},
]},
"rationale": "Art. 28 DSGVO.",
},
{
"rule_key": "vvt_required_l2plus",
"document_type": "vvt_register",
"title": "Verzeichnis von Verarbeitungstätigkeiten (VVT)",
"classification": "required",
"conditions": {"kind": "all", "clauses": [_level_clause("L2", "L3", "L4")]},
"rationale": "Art. 30 DSGVO — Pflicht außer Ausnahmen § 30 Abs. 5.",
},
{
"rule_key": "tom_required_l2plus",
"document_type": "tom_documentation",
"title": "Technisch-Organisatorische Maßnahmen (TOM)",
"classification": "required",
"conditions": {"kind": "all", "clauses": [_level_clause("L2", "L3", "L4")]},
"rationale": "Art. 32 DSGVO Nachweispflicht.",
},
{
"rule_key": "loeschkonzept_required_l2plus",
"document_type": "loeschkonzept",
"title": "Löschkonzept",
"classification": "required",
"conditions": {"kind": "all", "clauses": [_level_clause("L2", "L3", "L4")]},
"rationale": "Art. 5 (1) e + Art. 17 DSGVO.",
},
{
"rule_key": "tia_required_third_country",
"document_type": "transfer_impact_assessment",
"title": "Transfer Impact Assessment (TIA)",
"classification": "required",
"conditions": {"kind": "all", "clauses": [
{"field": "tech_third_country", "op": "not_in",
"value": ["no", "us_dpf_only", "adequate_only"]},
]},
"rationale": "EuGH C-311/18 (Schrems II) + EDSA-Empfehlungen 01/2020.",
},
{
"rule_key": "isms_required_when_certifying",
"document_type": "isms_manual",
"title": "ISMS-Handbuch",
"classification": "required",
"conditions": {"kind": "all", "clauses": [
{"field": "org_cert_target", "op": "in",
"value": ["iso27001", "iso27701", "tisax"]},
]},
"rationale": "ISO 27001 / TISAX VDA-ISA Pflichtdokumentation.",
},
{
"rule_key": "vendor_risk_recommended",
"document_type": "vendor_risk_management_policy",
"title": "Lieferantenrisiko-Management",
"classification": "recommended",
"conditions": {"kind": "any", "clauses": [
{"field": "comp_vendor_management", "op": "neq", "value": "no"},
{"field": "compliance_depth_level", "op": "eq", "value": "L4"},
]},
"rationale": "Empfehlenswert bei aktivem Vendor-Management.",
},
{
"rule_key": "bcm_required_l4",
"document_type": "business_continuity_policy",
"title": "Business-Continuity-Richtlinie",
"classification": "required",
"conditions": {"kind": "all", "clauses": [
{"field": "compliance_depth_level", "op": "eq", "value": "L4"},
]},
"rationale": "BSI 200-4 / ISO 22301 für Zertifizierungsstufe.",
},
]
__all__ = ["SEED_RULES"]
@@ -0,0 +1,137 @@
"""
SQLAlchemy models for Compliance Template Rules — profilbasierte
Empfehlungs-Regeln mit Versionierung, Approval-Workflow und Tenant-Overrides.
Tables:
- compliance_template_rules: Regel-Hülle
- compliance_template_rule_versions: Versionen mit Lifecycle
- compliance_template_rule_approvals: Audit-Trail
- compliance_tenant_rule_overrides: Pro-Tenant-Overrides globaler Regeln
"""
import uuid
from datetime import datetime
from sqlalchemy import (
Column, String, Text, SmallInteger, Integer, DateTime, Index, ForeignKey,
)
from sqlalchemy.dialects.postgresql import UUID, JSONB
from classroom_engine.database import Base
class TemplateRuleDB(Base):
"""Regel-Hülle: 1 Regel = 1 Empfehlung für ein document_type."""
__tablename__ = 'compliance_template_rules'
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
rule_key = Column(String(150), nullable=False, unique=True)
document_type = Column(String(100), nullable=False)
title = Column(String(300), nullable=False)
current_version_id = Column(UUID(as_uuid=True))
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
__table_args__ = (
Index('idx_template_rules_type', 'document_type'),
)
def __repr__(self) -> str:
return f"<TemplateRule {self.rule_key}{self.document_type}>"
class TemplateRuleVersionDB(Base):
"""Eine Version einer Regel mit Lifecycle (draft → review → approved → published)."""
__tablename__ = 'compliance_template_rule_versions'
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
rule_id = Column(
UUID(as_uuid=True),
ForeignKey('compliance_template_rules.id', ondelete='CASCADE'),
nullable=False,
)
version_number = Column(Integer, nullable=False)
status = Column(String(20), default='draft', nullable=False)
is_live = Column(SmallInteger, default=0, nullable=False)
classification = Column(String(20), nullable=False)
conditions = Column(JSONB, nullable=False, default=dict)
source_citation = Column(Text, nullable=False, default='')
rationale = Column(Text)
change_summary = Column(Text)
created_by = Column(String(200))
submitted_by = Column(String(200))
submitted_at = Column(DateTime)
approved_by = Column(String(200))
approved_at = Column(DateTime)
published_by = Column(String(200))
published_at = Column(DateTime)
rejected_by = Column(String(200))
rejected_at = Column(DateTime)
rejection_reason = Column(Text)
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
__table_args__ = (
Index('idx_rule_versions_rule', 'rule_id'),
Index('idx_rule_versions_status', 'status'),
)
def __repr__(self) -> str:
return f"<TemplateRuleVersion v{self.version_number} [{self.status}]>"
class TemplateRuleApprovalDB(Base):
"""Audit-Trail aller Lifecycle-Aktionen auf einer Regel-Version."""
__tablename__ = 'compliance_template_rule_approvals'
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
version_id = Column(
UUID(as_uuid=True),
ForeignKey('compliance_template_rule_versions.id', ondelete='CASCADE'),
nullable=False,
)
action = Column(String(50), nullable=False)
approver = Column(String(200))
comment = Column(Text)
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
__table_args__ = (
Index('idx_rule_approvals_version', 'version_id'),
)
def __repr__(self) -> str:
return f"<TemplateRuleApproval {self.action} on {self.version_id}>"
class TenantRuleOverrideDB(Base):
"""Override einer globalen Regel pro Tenant/Kanzlei."""
__tablename__ = 'compliance_tenant_rule_overrides'
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
tenant_id = Column(String(100), nullable=False)
rule_id = Column(
UUID(as_uuid=True),
ForeignKey('compliance_template_rules.id', ondelete='CASCADE'),
nullable=False,
)
override_classification = Column(String(20)) # null = deaktiviert
reason = Column(Text, nullable=False)
created_by = Column(String(200))
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
__table_args__ = (
Index('idx_tenant_overrides_tenant', 'tenant_id'),
)
def __repr__(self) -> str:
return f"<TenantRuleOverride tenant={self.tenant_id} rule={self.rule_id}>"
@@ -0,0 +1,174 @@
"""
Pydantic schemas for Compliance Template Rules — profilbasierte Empfehlungs-Regeln
mit Versionierung, Approval-Workflow und Tenant-Overrides.
"""
from datetime import datetime
from typing import Any, Optional
from pydantic import BaseModel, Field
# ----- Bedingungs-Strukturen -----
class RuleClause(BaseModel):
"""Eine einzelne Klausel innerhalb einer Bedingung."""
field: str
op: str # "eq" | "neq" | "gte" | "lte" | "gt" | "lt" | "in" | "not_in" | "exists"
value: Any = None
class RuleCondition(BaseModel):
"""Strukturierte Bedingung — kombiniert mehrere Klauseln per AND ('all') oder OR ('any')."""
kind: str = "all" # "all" oder "any"
clauses: list[RuleClause] = Field(default_factory=list)
# ----- Rule (Hülle) -----
class RuleCreate(BaseModel):
rule_key: str
document_type: str
title: str
class RuleResponse(BaseModel):
id: str
rule_key: str
document_type: str
title: str
current_version_id: Optional[str]
created_at: datetime
updated_at: Optional[datetime]
# ----- Version -----
class RuleVersionCreate(BaseModel):
"""Neuer Draft (entweder ganz neu oder fork einer Live-Version)."""
rule_id: str
classification: str # required | recommended | optional
conditions: RuleCondition
source_citation: str
rationale: Optional[str] = None
created_by: Optional[str] = None
class RuleVersionUpdate(BaseModel):
"""Edit eines Drafts — nur erlaubt während status='draft'."""
classification: Optional[str] = None
conditions: Optional[RuleCondition] = None
source_citation: Optional[str] = None
rationale: Optional[str] = None
change_summary: Optional[str] = None
class RuleVersionResponse(BaseModel):
id: str
rule_id: str
version_number: int
status: str
is_live: bool
classification: str
conditions: dict[str, Any]
source_citation: str
rationale: Optional[str]
change_summary: Optional[str]
created_by: Optional[str]
submitted_by: Optional[str]
submitted_at: Optional[datetime]
approved_by: Optional[str]
approved_at: Optional[datetime]
published_by: Optional[str]
published_at: Optional[datetime]
rejected_by: Optional[str]
rejected_at: Optional[datetime]
rejection_reason: Optional[str]
created_at: datetime
updated_at: Optional[datetime]
# ----- Lifecycle-Actions -----
class SubmitForReviewRequest(BaseModel):
change_summary: str
submitter: Optional[str] = None
comment: Optional[str] = None
class ApprovalActionRequest(BaseModel):
approver: Optional[str] = None
comment: Optional[str] = None
class RejectActionRequest(BaseModel):
rejector: Optional[str] = None
rejection_reason: str
comment: Optional[str] = None
class ApprovalHistoryEntry(BaseModel):
id: str
version_id: str
action: str
approver: Optional[str]
comment: Optional[str]
created_at: datetime
# ----- Tenant-Overrides -----
class OverrideCreate(BaseModel):
rule_id: str
override_classification: Optional[str] = None # null = deaktiviert
reason: str
created_by: Optional[str] = None
class OverrideResponse(BaseModel):
id: str
tenant_id: str
rule_id: str
override_classification: Optional[str]
reason: str
created_by: Optional[str]
created_at: datetime
updated_at: Optional[datetime]
# ----- Recommendation -----
class RecommendationRequest(BaseModel):
"""Input fürs Recommend: Profil-Antworten + optional Compliance-Tiefe."""
profile: dict[str, Any]
compliance_depth_level: Optional[str] = None # "L1"|"L2"|"L3"|"L4"
class RecommendedItem(BaseModel):
document_type: str
title: str
rule_id: str
rule_key: str
classification: str # required | recommended | optional (nach Override-Anwendung)
base_classification: str # ohne Override
source_citation: str
reason: str # menschenlesbare Begründung
override_applied: bool
class RecommendationResult(BaseModel):
required: list[RecommendedItem]
recommended: list[RecommendedItem]
optional: list[RecommendedItem]
profile_used: dict[str, Any]
__all__ = [
"RuleClause", "RuleCondition",
"RuleCreate", "RuleResponse",
"RuleVersionCreate", "RuleVersionUpdate", "RuleVersionResponse",
"SubmitForReviewRequest", "ApprovalActionRequest", "RejectActionRequest",
"ApprovalHistoryEntry",
"OverrideCreate", "OverrideResponse",
"RecommendationRequest", "RecommendedItem", "RecommendationResult",
]
@@ -0,0 +1,235 @@
# mypy: disable-error-code="arg-type,assignment,union-attr"
"""
Recommendation Service — Auswertung der Empfehlungs-Regeln gegen ein Profil.
- Lädt alle live (is_live=1, status='published') Regeln
- Lädt Tenant-Overrides
- Iteriert Regeln, evaluiert ``conditions`` JSONB gegen Profil-Dict
- Wendet Overrides an (Klassifikation überschreiben oder Regel deaktivieren)
- Gruppiert in required/recommended/optional mit Begründung
"""
from typing import Any, Optional
from sqlalchemy.orm import Session
from compliance.db.template_rule_models import (
TemplateRuleDB,
TemplateRuleVersionDB,
TenantRuleOverrideDB,
)
from compliance.schemas.template_rule import (
RecommendationRequest,
RecommendationResult,
RecommendedItem,
)
# ---------------------------------------------------------------- condition eval
def _eval_clause(clause: dict[str, Any], profile: dict[str, Any]) -> bool:
"""Evaluate eine einzelne Bedingungsklausel gegen das Profil-Dict."""
field = clause.get("field")
op = clause.get("op")
expected = clause.get("value")
if not field or not op:
return False
actual = profile.get(field)
if op == "exists":
return actual is not None and actual != "" and actual is not False
if op == "eq":
return actual == expected
if op == "neq":
return actual != expected
if op == "in":
if not isinstance(expected, list):
return False
return actual in expected
if op == "not_in":
if not isinstance(expected, list):
return True
return actual not in expected
if op in ("gt", "gte", "lt", "lte"):
try:
a = _to_number(actual)
e = _to_number(expected)
except (TypeError, ValueError):
return False
if op == "gt":
return a > e
if op == "gte":
return a >= e
if op == "lt":
return a < e
if op == "lte":
return a <= e
if op == "truthy":
return bool(actual)
if op == "falsy":
return not bool(actual)
return False
def _to_number(v: Any) -> float:
if isinstance(v, (int, float)):
return float(v)
if isinstance(v, str):
# Buckets wie "50_249" → nimm untere Schwelle
if "_" in v:
head = v.split("_")[0]
return float(head)
return float(v)
raise ValueError(f"Cannot coerce to number: {v}")
def _eval_condition(condition: dict[str, Any], profile: dict[str, Any]) -> bool:
"""Evaluate die gesamte Bedingung (all/any über Klauseln)."""
if not condition:
return True # leere Bedingung → immer wahr
kind = condition.get("kind", "all")
clauses = condition.get("clauses") or []
if not clauses:
return True
if kind == "all":
return all(_eval_clause(c, profile) for c in clauses)
if kind == "any":
return any(_eval_clause(c, profile) for c in clauses)
return False
def _build_reason(
rule: TemplateRuleDB,
version: TemplateRuleVersionDB,
profile: dict[str, Any],
override: Optional[TenantRuleOverrideDB],
) -> str:
"""Menschenlesbare Begründung für den Reviewer im Workspace."""
parts: list[str] = []
conds = version.conditions or {}
clauses = conds.get("clauses") or []
matched = [
c for c in clauses if _eval_clause(c, profile)
]
if matched:
bits = []
for c in matched[:3]:
field = c.get("field", "?")
op = c.get("op", "?")
val = c.get("value", "?")
actual = profile.get(field)
bits.append(f"{field} ({actual}) {op} {val}")
parts.append("Trifft zu: " + "; ".join(bits))
else:
parts.append(f"Always-rule für '{rule.document_type}'")
if version.source_citation:
parts.append(f"Quelle: {version.source_citation}")
if override:
if override.override_classification:
parts.append(
f"Tenant-Override: {version.classification}"
f"{override.override_classification} ({override.reason})"
)
else:
parts.append(f"Tenant-Override: deaktiviert ({override.reason})")
return " · ".join(parts)
# ---------------------------------------------------------------- service
class RecommendationService:
def __init__(self, db: Session) -> None:
self.db = db
def recommend(
self,
request: RecommendationRequest,
tenant_id: Optional[str] = None,
) -> RecommendationResult:
profile = dict(request.profile or {})
if request.compliance_depth_level:
profile.setdefault("compliance_depth_level", request.compliance_depth_level)
# Live-Versionen aller Regeln laden
live_versions = (
self.db.query(TemplateRuleVersionDB)
.filter(TemplateRuleVersionDB.is_live == 1)
.all()
)
if not live_versions:
return RecommendationResult(
required=[], recommended=[], optional=[], profile_used=profile,
)
rule_ids = [v.rule_id for v in live_versions]
rules = {
r.id: r for r in
self.db.query(TemplateRuleDB)
.filter(TemplateRuleDB.id.in_(rule_ids))
.all()
}
# Tenant-Overrides
overrides: dict[Any, TenantRuleOverrideDB] = {}
if tenant_id:
for o in (
self.db.query(TenantRuleOverrideDB)
.filter(TenantRuleOverrideDB.tenant_id == tenant_id)
.filter(TenantRuleOverrideDB.rule_id.in_(rule_ids))
.all()
):
overrides[o.rule_id] = o
buckets: dict[str, list[RecommendedItem]] = {
"required": [], "recommended": [], "optional": [],
}
for v in live_versions:
rule = rules.get(v.rule_id)
if not rule:
continue
if not _eval_condition(v.conditions or {}, profile):
continue
override = overrides.get(v.rule_id)
base_class = v.classification
effective_class = base_class
if override:
if override.override_classification is None:
# Override = deaktiviert
continue
effective_class = override.override_classification
reason = _build_reason(rule, v, profile, override)
item = RecommendedItem(
document_type=rule.document_type,
title=rule.title,
rule_id=str(rule.id),
rule_key=rule.rule_key,
classification=effective_class,
base_classification=base_class,
source_citation=v.source_citation or "",
reason=reason,
override_applied=override is not None,
)
buckets[effective_class].append(item)
# Sortiert nach document_type stabil
for k in buckets:
buckets[k].sort(key=lambda i: i.document_type)
return RecommendationResult(
required=buckets["required"],
recommended=buckets["recommended"],
optional=buckets["optional"],
profile_used=profile,
)
__all__ = ["RecommendationService"]
@@ -0,0 +1,499 @@
# mypy: disable-error-code="arg-type,assignment,union-attr"
"""
Template Rule service — Empfehlungs-Regeln + Versionierung + Approval-Workflow
+ Tenant-Overrides.
Analog zu ``legal_document_service.py`` aufgebaut:
- ``_transition()``-Helper für Lifecycle-Übergänge
- Audit-Trail in ``compliance_template_rule_approvals``
- Service-Klasse mit DB-Session-Konstruktor
"""
from datetime import datetime, timezone
from typing import Any, Optional
from sqlalchemy.orm import Session
from compliance.db.template_rule_models import (
TemplateRuleDB,
TemplateRuleVersionDB,
TemplateRuleApprovalDB,
TenantRuleOverrideDB,
)
from compliance.domain import NotFoundError, ValidationError
from compliance.schemas.template_rule import (
ApprovalActionRequest,
ApprovalHistoryEntry,
OverrideCreate,
OverrideResponse,
RejectActionRequest,
RuleCondition,
RuleCreate,
RuleResponse,
RuleVersionCreate,
RuleVersionResponse,
RuleVersionUpdate,
SubmitForReviewRequest,
)
# ---------------------------------------------------------------- response mappers
def _rule_to_response(rule: TemplateRuleDB) -> RuleResponse:
return RuleResponse(
id=str(rule.id),
rule_key=rule.rule_key,
document_type=rule.document_type,
title=rule.title,
current_version_id=str(rule.current_version_id) if rule.current_version_id else None,
created_at=rule.created_at,
updated_at=rule.updated_at,
)
def _version_to_response(v: TemplateRuleVersionDB) -> RuleVersionResponse:
return RuleVersionResponse(
id=str(v.id),
rule_id=str(v.rule_id),
version_number=v.version_number,
status=v.status,
is_live=bool(v.is_live),
classification=v.classification,
conditions=v.conditions or {},
source_citation=v.source_citation or "",
rationale=v.rationale,
change_summary=v.change_summary,
created_by=v.created_by,
submitted_by=v.submitted_by,
submitted_at=v.submitted_at,
approved_by=v.approved_by,
approved_at=v.approved_at,
published_by=v.published_by,
published_at=v.published_at,
rejected_by=v.rejected_by,
rejected_at=v.rejected_at,
rejection_reason=v.rejection_reason,
created_at=v.created_at,
updated_at=v.updated_at,
)
def _override_to_response(o: TenantRuleOverrideDB) -> OverrideResponse:
return OverrideResponse(
id=str(o.id),
tenant_id=o.tenant_id,
rule_id=str(o.rule_id),
override_classification=o.override_classification,
reason=o.reason,
created_by=o.created_by,
created_at=o.created_at,
updated_at=o.updated_at,
)
def _log_approval(
db: Session,
version_id: Any,
action: str,
approver: Optional[str] = None,
comment: Optional[str] = None,
) -> TemplateRuleApprovalDB:
entry = TemplateRuleApprovalDB(
version_id=version_id, action=action, approver=approver, comment=comment,
)
db.add(entry)
return entry
def _transition(
db: Session,
version_id: str,
from_statuses: list[str],
to_status: str,
action: str,
approver: Optional[str],
comment: Optional[str],
extra_updates: Optional[dict[str, Any]] = None,
) -> RuleVersionResponse:
version = (
db.query(TemplateRuleVersionDB)
.filter(TemplateRuleVersionDB.id == version_id)
.first()
)
if not version:
raise NotFoundError(f"Rule version {version_id} not found")
if version.status not in from_statuses:
raise ValidationError(
f"Cannot perform '{action}' on version with status "
f"'{version.status}' (expected: {from_statuses})"
)
version.status = to_status
version.updated_at = datetime.now(timezone.utc)
if extra_updates:
for k, v in extra_updates.items():
setattr(version, k, v)
_log_approval(db, version.id, action=action, approver=approver, comment=comment)
db.commit()
db.refresh(version)
return _version_to_response(version)
# ---------------------------------------------------------------- service
class TemplateRuleService:
"""Business logic for rules, versions, approval workflow and tenant overrides."""
def __init__(self, db: Session) -> None:
self.db = db
# ----- Rules (Hülle) -----
def list_rules(
self,
document_type: Optional[str] = None,
) -> list[RuleResponse]:
q = self.db.query(TemplateRuleDB)
if document_type:
q = q.filter(TemplateRuleDB.document_type == document_type)
rules = q.order_by(TemplateRuleDB.created_at.asc()).all()
return [_rule_to_response(r) for r in rules]
def get_rule(self, rule_id: str) -> RuleResponse:
rule = self._rule_or_raise(rule_id)
return _rule_to_response(rule)
def _rule_or_raise(self, rule_id: str) -> TemplateRuleDB:
rule = (
self.db.query(TemplateRuleDB)
.filter(TemplateRuleDB.id == rule_id)
.first()
)
if not rule:
raise NotFoundError(f"Rule {rule_id} not found")
return rule
def create_rule(self, request: RuleCreate) -> RuleResponse:
existing = (
self.db.query(TemplateRuleDB)
.filter(TemplateRuleDB.rule_key == request.rule_key)
.first()
)
if existing:
raise ValidationError(f"Rule with key '{request.rule_key}' already exists")
rule = TemplateRuleDB(
rule_key=request.rule_key,
document_type=request.document_type,
title=request.title,
)
self.db.add(rule)
self.db.commit()
self.db.refresh(rule)
return _rule_to_response(rule)
def delete_rule(self, rule_id: str) -> None:
rule = self._rule_or_raise(rule_id)
self.db.delete(rule)
self.db.commit()
# ----- Versions -----
def list_versions_for(self, rule_id: str) -> list[RuleVersionResponse]:
self._rule_or_raise(rule_id)
versions = (
self.db.query(TemplateRuleVersionDB)
.filter(TemplateRuleVersionDB.rule_id == rule_id)
.order_by(TemplateRuleVersionDB.version_number.desc())
.all()
)
return [_version_to_response(v) for v in versions]
def get_version(self, version_id: str) -> RuleVersionResponse:
v = self._version_or_raise(version_id)
return _version_to_response(v)
def _version_or_raise(self, version_id: str) -> TemplateRuleVersionDB:
v = (
self.db.query(TemplateRuleVersionDB)
.filter(TemplateRuleVersionDB.id == version_id)
.first()
)
if not v:
raise NotFoundError(f"Rule version {version_id} not found")
return v
def create_version(self, request: RuleVersionCreate) -> RuleVersionResponse:
"""Erzeugt einen neuen Draft. Vorhandener offener Draft blockiert."""
self._rule_or_raise(request.rule_id)
# Es darf nur eine nicht-finalisierte Version geben (draft|review)
open_v = (
self.db.query(TemplateRuleVersionDB)
.filter(TemplateRuleVersionDB.rule_id == request.rule_id)
.filter(TemplateRuleVersionDB.status.in_(["draft", "review"]))
.first()
)
if open_v:
raise ValidationError(
f"Rule {request.rule_id} already has an open draft (version {open_v.version_number})"
)
# Nächste Versionsnummer ermitteln
last = (
self.db.query(TemplateRuleVersionDB)
.filter(TemplateRuleVersionDB.rule_id == request.rule_id)
.order_by(TemplateRuleVersionDB.version_number.desc())
.first()
)
next_n = (last.version_number + 1) if last else 1
if request.classification not in ("required", "recommended", "optional"):
raise ValidationError(f"Invalid classification '{request.classification}'")
if not request.source_citation or not request.source_citation.strip():
raise ValidationError("source_citation is required (Pflichtfeld)")
version = TemplateRuleVersionDB(
rule_id=request.rule_id,
version_number=next_n,
status="draft",
is_live=0,
classification=request.classification,
conditions=request.conditions.dict(),
source_citation=request.source_citation.strip(),
rationale=request.rationale,
created_by=request.created_by,
)
self.db.add(version)
self.db.flush()
_log_approval(self.db, version.id, action="created", approver=request.created_by)
self.db.commit()
self.db.refresh(version)
return _version_to_response(version)
def update_version(
self, version_id: str, request: RuleVersionUpdate
) -> RuleVersionResponse:
v = self._version_or_raise(version_id)
if v.status != "draft":
raise ValidationError(
f"Only draft versions can be edited (current: {v.status})"
)
if request.classification is not None:
if request.classification not in ("required", "recommended", "optional"):
raise ValidationError(f"Invalid classification '{request.classification}'")
v.classification = request.classification
if request.conditions is not None:
v.conditions = request.conditions.dict()
if request.source_citation is not None:
v.source_citation = request.source_citation.strip()
if request.rationale is not None:
v.rationale = request.rationale
if request.change_summary is not None:
v.change_summary = request.change_summary
v.updated_at = datetime.now(timezone.utc)
self.db.commit()
self.db.refresh(v)
return _version_to_response(v)
# ----- Lifecycle -----
def submit_for_review(
self, version_id: str, request: SubmitForReviewRequest
) -> RuleVersionResponse:
if not request.change_summary or not request.change_summary.strip():
raise ValidationError("change_summary is required when submitting for review")
# change_summary erst persistieren
v = self._version_or_raise(version_id)
v.change_summary = request.change_summary.strip()
self.db.flush()
return _transition(
db=self.db,
version_id=version_id,
from_statuses=["draft"],
to_status="review",
action="submitted",
approver=request.submitter,
comment=request.comment,
extra_updates={
"submitted_by": request.submitter,
"submitted_at": datetime.now(timezone.utc),
},
)
def approve(
self, version_id: str, request: ApprovalActionRequest
) -> RuleVersionResponse:
return _transition(
db=self.db,
version_id=version_id,
from_statuses=["review"],
to_status="approved",
action="approved",
approver=request.approver,
comment=request.comment,
extra_updates={
"approved_by": request.approver,
"approved_at": datetime.now(timezone.utc),
},
)
def publish(
self, version_id: str, request: ApprovalActionRequest
) -> RuleVersionResponse:
# Auf 'approved' → publish: vorige published Version archivieren, aktuelle live machen
v = self._version_or_raise(version_id)
if v.status != "approved":
raise ValidationError(
f"Only approved versions can be published (current: {v.status})"
)
# Vorige live demoten
prev_live = (
self.db.query(TemplateRuleVersionDB)
.filter(TemplateRuleVersionDB.rule_id == v.rule_id)
.filter(TemplateRuleVersionDB.is_live == 1)
.first()
)
if prev_live:
prev_live.status = "archived"
prev_live.is_live = 0
prev_live.updated_at = datetime.now(timezone.utc)
_log_approval(self.db, prev_live.id, action="archived", approver=request.approver)
# Aktuelle live machen
now = datetime.now(timezone.utc)
v.status = "published"
v.is_live = 1
v.published_by = request.approver
v.published_at = now
v.updated_at = now
# current_version_id auf rule setzen
rule = self._rule_or_raise(str(v.rule_id))
rule.current_version_id = v.id
rule.updated_at = now
_log_approval(self.db, v.id, action="published", approver=request.approver, comment=request.comment)
self.db.commit()
self.db.refresh(v)
return _version_to_response(v)
def reject(
self, version_id: str, request: RejectActionRequest
) -> RuleVersionResponse:
if not request.rejection_reason or not request.rejection_reason.strip():
raise ValidationError("rejection_reason is required")
return _transition(
db=self.db,
version_id=version_id,
from_statuses=["review"],
to_status="rejected",
action="rejected",
approver=request.rejector,
comment=request.comment,
extra_updates={
"rejected_by": request.rejector,
"rejected_at": datetime.now(timezone.utc),
"rejection_reason": request.rejection_reason.strip(),
},
)
def approval_history(self, version_id: str) -> list[ApprovalHistoryEntry]:
self._version_or_raise(version_id)
entries = (
self.db.query(TemplateRuleApprovalDB)
.filter(TemplateRuleApprovalDB.version_id == version_id)
.order_by(TemplateRuleApprovalDB.created_at.asc())
.all()
)
return [
ApprovalHistoryEntry(
id=str(e.id),
version_id=str(e.version_id),
action=e.action,
approver=e.approver,
comment=e.comment,
created_at=e.created_at,
)
for e in entries
]
# ----- Tenant-Overrides -----
def list_overrides(self, tenant_id: str) -> list[OverrideResponse]:
os = (
self.db.query(TenantRuleOverrideDB)
.filter(TenantRuleOverrideDB.tenant_id == tenant_id)
.order_by(TenantRuleOverrideDB.created_at.desc())
.all()
)
return [_override_to_response(o) for o in os]
def upsert_override(
self, tenant_id: str, request: OverrideCreate
) -> OverrideResponse:
self._rule_or_raise(request.rule_id)
if request.override_classification is not None:
if request.override_classification not in ("required", "recommended", "optional"):
raise ValidationError(
f"Invalid override_classification '{request.override_classification}'"
)
if not request.reason or not request.reason.strip():
raise ValidationError("reason is required")
existing = (
self.db.query(TenantRuleOverrideDB)
.filter(TenantRuleOverrideDB.tenant_id == tenant_id)
.filter(TenantRuleOverrideDB.rule_id == request.rule_id)
.first()
)
if existing:
existing.override_classification = request.override_classification
existing.reason = request.reason.strip()
existing.created_by = request.created_by
existing.updated_at = datetime.now(timezone.utc)
self.db.commit()
self.db.refresh(existing)
return _override_to_response(existing)
override = TenantRuleOverrideDB(
tenant_id=tenant_id,
rule_id=request.rule_id,
override_classification=request.override_classification,
reason=request.reason.strip(),
created_by=request.created_by,
)
self.db.add(override)
self.db.commit()
self.db.refresh(override)
return _override_to_response(override)
def delete_override(self, tenant_id: str, override_id: str) -> None:
o = (
self.db.query(TenantRuleOverrideDB)
.filter(TenantRuleOverrideDB.id == override_id)
.filter(TenantRuleOverrideDB.tenant_id == tenant_id)
.first()
)
if not o:
raise NotFoundError(f"Override {override_id} not found for tenant {tenant_id}")
self.db.delete(o)
self.db.commit()
__all__ = [
"TemplateRuleService",
"_rule_to_response",
"_version_to_response",
"_override_to_response",
"_transition",
"_log_approval",
]
@@ -0,0 +1,106 @@
-- =========================================================
-- Migration 147: Compliance Template Rules — profilbasierte Empfehlungs-Regeln
-- mit Versionierung, Approval-Workflow und Tenant-Overrides.
--
-- Ersetzt langfristig die hartcodierten Regeln in
-- admin-compliance/app/sdk/document-generator/templateRecommendations.ts.
-- =========================================================
-- compliance_template_rules: Regel-Hülle (eine Regel = eine document_type-Empfehlung)
CREATE TABLE IF NOT EXISTS compliance_template_rules (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
rule_key VARCHAR(150) NOT NULL UNIQUE, -- z.B. "whistleblower_required_50plus"
document_type VARCHAR(100) NOT NULL, -- z.B. "whistleblower_policy"
title VARCHAR(300) NOT NULL, -- kurzer Anzeigename
current_version_id UUID, -- pointet auf published Version, null bis erste Freigabe
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
-- compliance_template_rule_versions: Versionen mit Lifecycle + Approval-Trail
CREATE TABLE IF NOT EXISTS compliance_template_rule_versions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
rule_id UUID NOT NULL REFERENCES compliance_template_rules(id) ON DELETE CASCADE,
version_number INT NOT NULL,
-- Lifecycle (analog compliance_legal_document_versions)
status VARCHAR(20) NOT NULL DEFAULT 'draft', -- draft|review|approved|published|archived|rejected
is_live SMALLINT NOT NULL DEFAULT 0, -- 1 = aktuell published Version dieser Regel
-- Klassifikation der Empfehlung
classification VARCHAR(20) NOT NULL, -- required | recommended | optional
-- Bedingung als strukturiertes JSONB
-- Format: { "kind": "all"|"any", "clauses": [{ "field": "...", "op": "...", "value": ... }] }
conditions JSONB NOT NULL DEFAULT '{"kind":"all","clauses":[]}'::jsonb,
-- Editorial-Pflichtfelder
source_citation TEXT NOT NULL DEFAULT '', -- "§ 12 HinSchG", "EuGH C-311/18", "DSK Kurzpapier Nr. 18"
rationale TEXT, -- Freitext-Begründung des Anwalts
change_summary TEXT, -- Pflicht beim Einreichen (was wurde geändert)
-- Approval-Trail (vereinfacht; volle Action-Liste in approvals-Tabelle)
created_by VARCHAR(200),
submitted_by VARCHAR(200),
submitted_at TIMESTAMPTZ,
approved_by VARCHAR(200),
approved_at TIMESTAMPTZ,
published_by VARCHAR(200),
published_at TIMESTAMPTZ,
rejected_by VARCHAR(200),
rejected_at TIMESTAMPTZ,
rejection_reason TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE (rule_id, version_number),
CHECK (classification IN ('required', 'recommended', 'optional')),
CHECK (status IN ('draft', 'review', 'approved', 'published', 'archived', 'rejected')),
CHECK (is_live IN (0, 1))
);
-- Genau eine published/live Version pro Regel
CREATE UNIQUE INDEX IF NOT EXISTS idx_rule_versions_one_live
ON compliance_template_rule_versions(rule_id) WHERE is_live = 1;
-- compliance_template_rule_approvals: Audit-Trail aller Lifecycle-Aktionen
CREATE TABLE IF NOT EXISTS compliance_template_rule_approvals (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
version_id UUID NOT NULL REFERENCES compliance_template_rule_versions(id) ON DELETE CASCADE,
action VARCHAR(50) NOT NULL, -- submitted|approved|rejected|published|archived
approver VARCHAR(200),
comment TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
-- compliance_tenant_rule_overrides: Pro Kanzlei/Tenant Override einer globalen Regel
CREATE TABLE IF NOT EXISTS compliance_tenant_rule_overrides (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id VARCHAR(100) NOT NULL,
rule_id UUID NOT NULL REFERENCES compliance_template_rules(id) ON DELETE CASCADE,
override_classification VARCHAR(20), -- null = Regel komplett für Tenant deaktiviert
reason TEXT NOT NULL, -- Pflicht: warum dieser Override
created_by VARCHAR(200),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
CONSTRAINT chk_override_classification CHECK (
override_classification IS NULL OR
override_classification IN ('required', 'recommended', 'optional')
)
);
-- Indizes
CREATE INDEX IF NOT EXISTS idx_template_rules_type
ON compliance_template_rules(document_type);
CREATE INDEX IF NOT EXISTS idx_rule_versions_rule
ON compliance_template_rule_versions(rule_id);
CREATE INDEX IF NOT EXISTS idx_rule_versions_status
ON compliance_template_rule_versions(status);
CREATE INDEX IF NOT EXISTS idx_rule_approvals_version
ON compliance_template_rule_approvals(version_id);
CREATE INDEX IF NOT EXISTS idx_tenant_overrides_tenant
ON compliance_tenant_rule_overrides(tenant_id);
CREATE UNIQUE INDEX IF NOT EXISTS idx_tenant_overrides_unique
ON compliance_tenant_rule_overrides(tenant_id, rule_id);
@@ -0,0 +1,89 @@
#!/usr/bin/env python3
"""
Seed-Runner: legt die in ``compliance.data.template_rule_seed_data.SEED_RULES``
definierten Regeln als published Versionen in ``compliance_template_rules`` an.
Idempotent über ``rule_key`` — wiederholtes Ausführen erzeugt keine Duplikate.
Quellen-Citation = ``rationale`` als Default (anwaltlich zu prüfen).
"""
from __future__ import annotations
import os
import sys
from datetime import datetime, timezone
# Pfad-Setup für Standalone-Ausführung
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from classroom_engine.database import SessionLocal # noqa: E402
from compliance.data.template_rule_seed_data import SEED_RULES # noqa: E402
from compliance.db.template_rule_models import ( # noqa: E402
TemplateRuleDB,
TemplateRuleVersionDB,
TemplateRuleApprovalDB,
)
def seed() -> None:
db = SessionLocal()
try:
created = 0
skipped = 0
for entry in SEED_RULES:
existing = (
db.query(TemplateRuleDB)
.filter(TemplateRuleDB.rule_key == entry["rule_key"])
.first()
)
if existing:
skipped += 1
continue
rule = TemplateRuleDB(
rule_key=entry["rule_key"],
document_type=entry["document_type"],
title=entry["title"],
)
db.add(rule)
db.flush()
now = datetime.now(timezone.utc)
version = TemplateRuleVersionDB(
rule_id=rule.id,
version_number=1,
status="published",
is_live=1,
classification=entry["classification"],
conditions=entry["conditions"],
source_citation=entry.get("rationale", "TODO — anwaltlich zu pruefen"),
rationale=entry.get("rationale"),
change_summary="Initial-Seed der Inline-Regeln",
created_by="seed",
submitted_by="seed",
submitted_at=now,
approved_by="seed",
approved_at=now,
published_by="seed",
published_at=now,
)
db.add(version)
db.flush()
rule.current_version_id = version.id
for action in ("created", "submitted", "approved", "published"):
db.add(TemplateRuleApprovalDB(
version_id=version.id, action=action, approver="seed",
))
created += 1
db.commit()
print(f"OK created={created} skipped={skipped} (already present)")
finally:
db.close()
if __name__ == "__main__":
seed()