refactor(backend/api): extract SourcePolicyService (Step 4 — file 7 of 18)

compliance/api/source_policy_router.py (580 LOC) -> 253 LOC thin routes
+ 453-line SourcePolicyService + 83-line schemas file. Manages allowed
data sources, operations matrix, PII rules, blocked-content log,
audit trail, and dashboard stats/report.

Single-service split. ORM-based (uses compliance.db.source_policy_models).

Date-string parsing extracted to a module-level _parse_iso_optional
helper so the audit + blocked-content list endpoints share it instead
of duplicating try/except blocks.

Legacy test compat: SourceCreate, SourceUpdate, SourceResponse,
PIIRuleCreate, PIIRuleUpdate, OperationUpdate, _log_audit re-exported
from compliance.api.source_policy_router via __all__.

Verified:
  - 208/208 pytest pass (173 core + 35 source policy)
  - OpenAPI 360/484 unchanged
  - mypy compliance/ -> Success on 132 source files
  - source_policy_router.py 580 -> 253 LOC
  - Hard-cap violations: 12 -> 11

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sharang Parnerkar
2026-04-07 19:58:02 +02:00
parent b850368ec9
commit 7107a31496
5 changed files with 740 additions and 469 deletions

View File

@@ -1,142 +1,56 @@
"""
Source Policy Router — Manages allowed compliance data sources.
Controls which legal sources the RAG corpus may use,
operations matrix, PII rules, and provides audit trail.
Controls which legal sources the RAG corpus may use, the operations matrix,
PII rules, blocked-content log, audit trail, and dashboard stats/report.
Endpoints:
GET /api/v1/admin/sources List all sources
POST /api/v1/admin/sources Add new source
GET /api/v1/admin/sources/{id} Get source by ID
PUT /api/v1/admin/sources/{id} Update source
DELETE /api/v1/admin/sources/{id} Remove source
GET /api/v1/admin/operations-matrix Operations matrix
PUT /api/v1/admin/operations/{id} Update operation
GET /api/v1/admin/pii-rules List PII rules
POST /api/v1/admin/pii-rules Create PII rule
PUT /api/v1/admin/pii-rules/{id} Update PII rule
DELETE /api/v1/admin/pii-rules/{id} Delete PII rule
GET /api/v1/admin/policy-audit — Audit trail
GET /api/v1/admin/policy-stats — Dashboard statistics
GET /api/v1/admin/compliance-report — Compliance report
GET /v1/admin/sources - List all sources
POST /v1/admin/sources - Add new source
GET /v1/admin/sources/{id} - Get source by ID
PUT /v1/admin/sources/{id} - Update source
DELETE /v1/admin/sources/{id} - Remove source
GET /v1/admin/operations-matrix - Operations matrix
PUT /v1/admin/operations/{id} - Update operation
GET /v1/admin/pii-rules - List PII rules
POST /v1/admin/pii-rules - Create PII rule
PUT /v1/admin/pii-rules/{id} - Update PII rule
DELETE /v1/admin/pii-rules/{id} - Delete PII rule
GET /v1/admin/blocked-content - Blocked content log
GET /v1/admin/policy-audit - Audit trail
GET /v1/admin/policy-stats - Dashboard statistics
GET /v1/admin/compliance-report - Compliance report
Phase 1 Step 4 refactor: handlers delegate to SourcePolicyService.
"""
from datetime import datetime, timezone
from typing import Optional
from typing import Any, Optional
from fastapi import APIRouter, HTTPException, Depends, Query
from pydantic import BaseModel, ConfigDict, Field
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from database import get_db
from compliance.db.source_policy_models import (
AllowedSourceDB,
BlockedContentDB,
SourceOperationDB,
PIIRuleDB,
SourcePolicyAuditDB,
from compliance.api._http_errors import translate_domain_errors
from compliance.schemas.source_policy import (
OperationUpdate,
PIIRuleCreate,
PIIRuleUpdate,
SourceCreate,
SourceResponse,
SourceUpdate,
)
from compliance.services.source_policy_service import (
SourcePolicyService,
_log_audit, # re-exported for legacy test imports
)
router = APIRouter(prefix="/v1/admin", tags=["source-policy"])
# =============================================================================
# Pydantic Schemas
# =============================================================================
class SourceCreate(BaseModel):
domain: str
name: str
description: Optional[str] = None
license: Optional[str] = None
legal_basis: Optional[str] = None
trust_boost: float = Field(default=0.5, ge=0.0, le=1.0)
source_type: str = "legal"
active: bool = True
metadata: Optional[dict] = None
class SourceUpdate(BaseModel):
domain: Optional[str] = None
name: Optional[str] = None
description: Optional[str] = None
license: Optional[str] = None
legal_basis: Optional[str] = None
trust_boost: Optional[float] = Field(default=None, ge=0.0, le=1.0)
source_type: Optional[str] = None
active: Optional[bool] = None
metadata: Optional[dict] = None
class SourceResponse(BaseModel):
id: str
domain: str
name: str
description: Optional[str] = None
license: Optional[str] = None
legal_basis: Optional[str] = None
trust_boost: float
source_type: str
active: bool
metadata: Optional[dict] = None
created_at: str
updated_at: Optional[str] = None
model_config = ConfigDict(from_attributes=True)
class OperationUpdate(BaseModel):
allowed: bool
conditions: Optional[str] = None
class PIIRuleCreate(BaseModel):
name: str
description: Optional[str] = None
pattern: Optional[str] = None
category: str
action: str = "mask"
active: bool = True
class PIIRuleUpdate(BaseModel):
name: Optional[str] = None
description: Optional[str] = None
pattern: Optional[str] = None
category: Optional[str] = None
action: Optional[str] = None
active: Optional[bool] = None
# =============================================================================
# Helper: Audit logging
# =============================================================================
def _log_audit(db: Session, action: str, entity_type: str, entity_id, old_values=None, new_values=None):
audit = SourcePolicyAuditDB(
action=action,
entity_type=entity_type,
entity_id=entity_id,
old_values=old_values,
new_values=new_values,
user_id="system",
)
db.add(audit)
def _source_to_dict(source: AllowedSourceDB) -> dict:
return {
"id": str(source.id),
"domain": source.domain,
"name": source.name,
"description": source.description,
"license": source.license,
"legal_basis": source.legal_basis,
"trust_boost": source.trust_boost,
"source_type": source.source_type,
"active": source.active,
}
def get_source_policy_service(
db: Session = Depends(get_db),
) -> SourcePolicyService:
return SourcePolicyService(db)
# =============================================================================
@@ -148,139 +62,52 @@ async def list_sources(
active_only: bool = Query(False),
source_type: Optional[str] = Query(None),
license: Optional[str] = Query(None),
db: Session = Depends(get_db),
):
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""List all allowed sources with optional filters."""
query = db.query(AllowedSourceDB)
if active_only:
query = query.filter(AllowedSourceDB.active)
if source_type:
query = query.filter(AllowedSourceDB.source_type == source_type)
if license:
query = query.filter(AllowedSourceDB.license == license)
sources = query.order_by(AllowedSourceDB.name).all()
return {
"sources": [
{
"id": str(s.id),
"domain": s.domain,
"name": s.name,
"description": s.description,
"license": s.license,
"legal_basis": s.legal_basis,
"trust_boost": s.trust_boost,
"source_type": s.source_type,
"active": s.active,
"metadata": s.metadata_,
"created_at": s.created_at.isoformat() if s.created_at else None,
"updated_at": s.updated_at.isoformat() if s.updated_at else None,
}
for s in sources
],
"count": len(sources),
}
with translate_domain_errors():
return service.list_sources(active_only, source_type, license)
@router.post("/sources")
async def create_source(
data: SourceCreate,
db: Session = Depends(get_db),
):
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Add a new allowed source."""
existing = db.query(AllowedSourceDB).filter(AllowedSourceDB.domain == data.domain).first()
if existing:
raise HTTPException(status_code=409, detail=f"Source with domain '{data.domain}' already exists")
source = AllowedSourceDB(
domain=data.domain,
name=data.name,
description=data.description,
license=data.license,
legal_basis=data.legal_basis,
trust_boost=data.trust_boost,
source_type=data.source_type,
active=data.active,
metadata_=data.metadata,
)
db.add(source)
_log_audit(db, "create", "source", source.id, new_values=_source_to_dict(source))
db.commit()
db.refresh(source)
return {
"id": str(source.id),
"domain": source.domain,
"name": source.name,
"created_at": source.created_at.isoformat(),
}
with translate_domain_errors():
return service.create_source(data)
@router.get("/sources/{source_id}")
async def get_source(source_id: str, db: Session = Depends(get_db)):
async def get_source(
source_id: str,
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Get a specific source."""
source = db.query(AllowedSourceDB).filter(AllowedSourceDB.id == source_id).first()
if not source:
raise HTTPException(status_code=404, detail="Source not found")
return {
"id": str(source.id),
"domain": source.domain,
"name": source.name,
"description": source.description,
"license": source.license,
"legal_basis": source.legal_basis,
"trust_boost": source.trust_boost,
"source_type": source.source_type,
"active": source.active,
"metadata": source.metadata_,
"created_at": source.created_at.isoformat() if source.created_at else None,
"updated_at": source.updated_at.isoformat() if source.updated_at else None,
}
with translate_domain_errors():
return service.get_source(source_id)
@router.put("/sources/{source_id}")
async def update_source(
source_id: str,
data: SourceUpdate,
db: Session = Depends(get_db),
):
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Update an existing source."""
source = db.query(AllowedSourceDB).filter(AllowedSourceDB.id == source_id).first()
if not source:
raise HTTPException(status_code=404, detail="Source not found")
old_values = _source_to_dict(source)
update_data = data.model_dump(exclude_unset=True)
# Rename metadata to metadata_ for the DB column
if "metadata" in update_data:
update_data["metadata_"] = update_data.pop("metadata")
for key, value in update_data.items():
setattr(source, key, value)
_log_audit(db, "update", "source", source.id, old_values=old_values, new_values=update_data)
db.commit()
db.refresh(source)
return {"status": "updated", "id": str(source.id)}
with translate_domain_errors():
return service.update_source(source_id, data)
@router.delete("/sources/{source_id}")
async def delete_source(source_id: str, db: Session = Depends(get_db)):
async def delete_source(
source_id: str,
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Remove an allowed source."""
source = db.query(AllowedSourceDB).filter(AllowedSourceDB.id == source_id).first()
if not source:
raise HTTPException(status_code=404, detail="Source not found")
old_values = _source_to_dict(source)
_log_audit(db, "delete", "source", source.id, old_values=old_values)
# Also delete associated operations
db.query(SourceOperationDB).filter(SourceOperationDB.source_id == source_id).delete()
db.delete(source)
db.commit()
return {"status": "deleted", "id": source_id}
with translate_domain_errors():
return service.delete_source(source_id)
# =============================================================================
@@ -288,43 +115,23 @@ async def delete_source(source_id: str, db: Session = Depends(get_db)):
# =============================================================================
@router.get("/operations-matrix")
async def get_operations_matrix(db: Session = Depends(get_db)):
async def get_operations_matrix(
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Get the full operations matrix."""
operations = db.query(SourceOperationDB).all()
return {
"operations": [
{
"id": str(op.id),
"source_id": str(op.source_id),
"operation": op.operation,
"allowed": op.allowed,
"conditions": op.conditions,
}
for op in operations
],
"count": len(operations),
}
with translate_domain_errors():
return service.get_operations_matrix()
@router.put("/operations/{operation_id}")
async def update_operation(
operation_id: str,
data: OperationUpdate,
db: Session = Depends(get_db),
):
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Update an operation in the matrix."""
op = db.query(SourceOperationDB).filter(SourceOperationDB.id == operation_id).first()
if not op:
raise HTTPException(status_code=404, detail="Operation not found")
op.allowed = data.allowed
if data.conditions is not None:
op.conditions = data.conditions
_log_audit(db, "update", "operation", op.id, new_values={"allowed": data.allowed})
db.commit()
return {"status": "updated", "id": str(op.id)}
with translate_domain_errors():
return service.update_operation(operation_id, data)
# =============================================================================
@@ -334,79 +141,42 @@ async def update_operation(
@router.get("/pii-rules")
async def list_pii_rules(
category: Optional[str] = Query(None),
db: Session = Depends(get_db),
):
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""List all PII rules with optional category filter."""
query = db.query(PIIRuleDB)
if category:
query = query.filter(PIIRuleDB.category == category)
rules = query.order_by(PIIRuleDB.category, PIIRuleDB.name).all()
return {
"rules": [
{
"id": str(r.id),
"name": r.name,
"description": r.description,
"pattern": r.pattern,
"category": r.category,
"action": r.action,
"active": r.active,
"created_at": r.created_at.isoformat() if r.created_at else None,
}
for r in rules
],
"count": len(rules),
}
with translate_domain_errors():
return service.list_pii_rules(category)
@router.post("/pii-rules")
async def create_pii_rule(data: PIIRuleCreate, db: Session = Depends(get_db)):
async def create_pii_rule(
data: PIIRuleCreate,
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Create a new PII rule."""
rule = PIIRuleDB(
name=data.name,
description=data.description,
pattern=data.pattern,
category=data.category,
action=data.action,
active=data.active,
)
db.add(rule)
_log_audit(db, "create", "pii_rule", rule.id, new_values={"name": data.name, "category": data.category})
db.commit()
db.refresh(rule)
return {"id": str(rule.id), "name": rule.name}
with translate_domain_errors():
return service.create_pii_rule(data)
@router.put("/pii-rules/{rule_id}")
async def update_pii_rule(rule_id: str, data: PIIRuleUpdate, db: Session = Depends(get_db)):
async def update_pii_rule(
rule_id: str,
data: PIIRuleUpdate,
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Update a PII rule."""
rule = db.query(PIIRuleDB).filter(PIIRuleDB.id == rule_id).first()
if not rule:
raise HTTPException(status_code=404, detail="PII rule not found")
update_data = data.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(rule, key, value)
_log_audit(db, "update", "pii_rule", rule.id, new_values=update_data)
db.commit()
return {"status": "updated", "id": str(rule.id)}
with translate_domain_errors():
return service.update_pii_rule(rule_id, data)
@router.delete("/pii-rules/{rule_id}")
async def delete_pii_rule(rule_id: str, db: Session = Depends(get_db)):
async def delete_pii_rule(
rule_id: str,
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Delete a PII rule."""
rule = db.query(PIIRuleDB).filter(PIIRuleDB.id == rule_id).first()
if not rule:
raise HTTPException(status_code=404, detail="PII rule not found")
_log_audit(db, "delete", "pii_rule", rule.id, old_values={"name": rule.name, "category": rule.category})
db.delete(rule)
db.commit()
return {"status": "deleted", "id": rule_id}
with translate_domain_errors():
return service.delete_pii_rule(rule_id)
# =============================================================================
@@ -420,46 +190,11 @@ async def list_blocked_content(
domain: Optional[str] = None,
date_from: Optional[str] = Query(None, alias="from"),
date_to: Optional[str] = Query(None, alias="to"),
db: Session = Depends(get_db),
):
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""List blocked content entries."""
query = db.query(BlockedContentDB)
if domain:
query = query.filter(BlockedContentDB.domain == domain)
if date_from:
try:
from_dt = datetime.fromisoformat(date_from)
query = query.filter(BlockedContentDB.created_at >= from_dt)
except ValueError:
pass
if date_to:
try:
to_dt = datetime.fromisoformat(date_to)
query = query.filter(BlockedContentDB.created_at <= to_dt)
except ValueError:
pass
total = query.count()
entries = query.order_by(BlockedContentDB.created_at.desc()).offset(offset).limit(limit).all()
return {
"blocked": [
{
"id": str(e.id),
"url": e.url,
"domain": e.domain,
"block_reason": e.block_reason,
"rule_id": str(e.rule_id) if e.rule_id else None,
"details": e.details,
"created_at": e.created_at.isoformat() if e.created_at else None,
}
for e in entries
],
"total": total,
}
with translate_domain_errors():
return service.list_blocked_content(limit, offset, domain, date_from, date_to)
# =============================================================================
@@ -473,108 +208,46 @@ async def get_policy_audit(
entity_type: Optional[str] = None,
date_from: Optional[str] = Query(None, alias="from"),
date_to: Optional[str] = Query(None, alias="to"),
db: Session = Depends(get_db),
):
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Get the audit trail for source policy changes."""
query = db.query(SourcePolicyAuditDB)
if entity_type:
query = query.filter(SourcePolicyAuditDB.entity_type == entity_type)
if date_from:
try:
from_dt = datetime.fromisoformat(date_from)
query = query.filter(SourcePolicyAuditDB.created_at >= from_dt)
except ValueError:
pass
if date_to:
try:
to_dt = datetime.fromisoformat(date_to)
query = query.filter(SourcePolicyAuditDB.created_at <= to_dt)
except ValueError:
pass
total = query.count()
entries = query.order_by(SourcePolicyAuditDB.created_at.desc()).offset(offset).limit(limit).all()
return {
"entries": [
{
"id": str(e.id),
"action": e.action,
"entity_type": e.entity_type,
"entity_id": str(e.entity_id) if e.entity_id else None,
"old_values": e.old_values,
"new_values": e.new_values,
"user_id": e.user_id,
"created_at": e.created_at.isoformat() if e.created_at else None,
}
for e in entries
],
"total": total,
"limit": limit,
"offset": offset,
}
with translate_domain_errors():
return service.get_audit(limit, offset, entity_type, date_from, date_to)
# =============================================================================
# Dashboard Statistics
# Dashboard Statistics + Report
# =============================================================================
@router.get("/policy-stats")
async def get_policy_stats(db: Session = Depends(get_db)):
async def get_policy_stats(
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Get dashboard statistics for source policy."""
total_sources = db.query(AllowedSourceDB).count()
active_sources = db.query(AllowedSourceDB).filter(AllowedSourceDB.active).count()
pii_rules = db.query(PIIRuleDB).filter(PIIRuleDB.active).count()
# Count blocked content entries from today
today_start = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
blocked_today = db.query(BlockedContentDB).filter(
BlockedContentDB.created_at >= today_start,
).count()
blocked_total = db.query(BlockedContentDB).count()
return {
"active_policies": active_sources,
"allowed_sources": total_sources,
"pii_rules": pii_rules,
"blocked_today": blocked_today,
"blocked_total": blocked_total,
}
with translate_domain_errors():
return service.stats()
@router.get("/compliance-report")
async def get_compliance_report(db: Session = Depends(get_db)):
async def get_compliance_report(
service: SourcePolicyService = Depends(get_source_policy_service),
) -> dict[str, Any]:
"""Generate a compliance report for source policies."""
sources = db.query(AllowedSourceDB).filter(AllowedSourceDB.active).all()
pii_rules = db.query(PIIRuleDB).filter(PIIRuleDB.active).all()
with translate_domain_errors():
return service.compliance_report()
return {
"report_date": datetime.now(timezone.utc).isoformat(),
"summary": {
"active_sources": len(sources),
"active_pii_rules": len(pii_rules),
"source_types": list(set(s.source_type for s in sources)),
"licenses": list(set(s.license for s in sources if s.license)),
},
"sources": [
{
"domain": s.domain,
"name": s.name,
"license": s.license,
"legal_basis": s.legal_basis,
"trust_boost": s.trust_boost,
}
for s in sources
],
"pii_rules": [
{
"name": r.name,
"category": r.category,
"action": r.action,
}
for r in pii_rules
],
}
# ----------------------------------------------------------------------------
# Legacy re-exports for tests that import schemas/helpers directly.
# ----------------------------------------------------------------------------
__all__ = [
"router",
"SourceCreate",
"SourceUpdate",
"SourceResponse",
"OperationUpdate",
"PIIRuleCreate",
"PIIRuleUpdate",
"_log_audit",
]

View File

@@ -0,0 +1,83 @@
"""
Source Policy schemas — allowed source registry, operations matrix, PII rules.
Phase 1 Step 4: extracted from ``compliance.api.source_policy_router``.
"""
from typing import Any, Optional
from pydantic import BaseModel, ConfigDict, Field
class SourceCreate(BaseModel):
domain: str
name: str
description: Optional[str] = None
license: Optional[str] = None
legal_basis: Optional[str] = None
trust_boost: float = Field(default=0.5, ge=0.0, le=1.0)
source_type: str = "legal"
active: bool = True
metadata: Optional[dict[str, Any]] = None
class SourceUpdate(BaseModel):
domain: Optional[str] = None
name: Optional[str] = None
description: Optional[str] = None
license: Optional[str] = None
legal_basis: Optional[str] = None
trust_boost: Optional[float] = Field(default=None, ge=0.0, le=1.0)
source_type: Optional[str] = None
active: Optional[bool] = None
metadata: Optional[dict[str, Any]] = None
class SourceResponse(BaseModel):
id: str
domain: str
name: str
description: Optional[str] = None
license: Optional[str] = None
legal_basis: Optional[str] = None
trust_boost: float
source_type: str
active: bool
metadata: Optional[dict[str, Any]] = None
created_at: str
updated_at: Optional[str] = None
model_config = ConfigDict(from_attributes=True)
class OperationUpdate(BaseModel):
allowed: bool
conditions: Optional[str] = None
class PIIRuleCreate(BaseModel):
name: str
description: Optional[str] = None
pattern: Optional[str] = None
category: str
action: str = "mask"
active: bool = True
class PIIRuleUpdate(BaseModel):
name: Optional[str] = None
description: Optional[str] = None
pattern: Optional[str] = None
category: Optional[str] = None
action: Optional[str] = None
active: Optional[bool] = None
__all__ = [
"SourceCreate",
"SourceUpdate",
"SourceResponse",
"OperationUpdate",
"PIIRuleCreate",
"PIIRuleUpdate",
]

View File

@@ -0,0 +1,453 @@
# mypy: disable-error-code="arg-type,assignment,union-attr"
"""
Source Policy service — allowed sources, operations matrix, PII rules,
blocked content, audit, stats, compliance report.
Phase 1 Step 4: extracted from ``compliance.api.source_policy_router``.
"""
from datetime import datetime, timezone
from typing import Any, Optional
from sqlalchemy.orm import Session
from compliance.db.source_policy_models import (
AllowedSourceDB,
BlockedContentDB,
PIIRuleDB,
SourceOperationDB,
SourcePolicyAuditDB,
)
from compliance.domain import ConflictError, NotFoundError
from compliance.schemas.source_policy import (
OperationUpdate,
PIIRuleCreate,
PIIRuleUpdate,
SourceCreate,
SourceUpdate,
)
# ============================================================================
# Module-level helpers (re-exported by compliance.api.source_policy_router for
# legacy test imports).
# ============================================================================
def _log_audit(
db: Session,
action: str,
entity_type: str,
entity_id: Any,
old_values: Optional[dict[str, Any]] = None,
new_values: Optional[dict[str, Any]] = None,
) -> None:
db.add(
SourcePolicyAuditDB(
action=action,
entity_type=entity_type,
entity_id=entity_id,
old_values=old_values,
new_values=new_values,
user_id="system",
)
)
def _source_to_dict(s: AllowedSourceDB) -> dict[str, Any]:
return {
"id": str(s.id),
"domain": s.domain,
"name": s.name,
"description": s.description,
"license": s.license,
"legal_basis": s.legal_basis,
"trust_boost": s.trust_boost,
"source_type": s.source_type,
"active": s.active,
}
def _full_source_dict(s: AllowedSourceDB) -> dict[str, Any]:
return {
**_source_to_dict(s),
"metadata": s.metadata_,
"created_at": s.created_at.isoformat() if s.created_at else None,
"updated_at": s.updated_at.isoformat() if s.updated_at else None,
}
def _parse_iso_optional(value: Optional[str]) -> Optional[datetime]:
if not value:
return None
try:
return datetime.fromisoformat(value)
except ValueError:
return None
# ============================================================================
# Service
# ============================================================================
class SourcePolicyService:
"""Business logic for the source policy admin surface."""
def __init__(self, db: Session) -> None:
self.db = db
# ------------------------------------------------------------------
# Sources CRUD
# ------------------------------------------------------------------
def list_sources(
self,
active_only: bool,
source_type: Optional[str],
license: Optional[str],
) -> dict[str, Any]:
q = self.db.query(AllowedSourceDB)
if active_only:
q = q.filter(AllowedSourceDB.active)
if source_type:
q = q.filter(AllowedSourceDB.source_type == source_type)
if license:
q = q.filter(AllowedSourceDB.license == license)
sources = q.order_by(AllowedSourceDB.name).all()
return {
"sources": [_full_source_dict(s) for s in sources],
"count": len(sources),
}
def create_source(self, data: SourceCreate) -> dict[str, Any]:
existing = (
self.db.query(AllowedSourceDB)
.filter(AllowedSourceDB.domain == data.domain)
.first()
)
if existing:
raise ConflictError(
f"Source with domain '{data.domain}' already exists"
)
source = AllowedSourceDB(
domain=data.domain,
name=data.name,
description=data.description,
license=data.license,
legal_basis=data.legal_basis,
trust_boost=data.trust_boost,
source_type=data.source_type,
active=data.active,
metadata_=data.metadata,
)
self.db.add(source)
_log_audit(
self.db, "create", "source", source.id,
new_values=_source_to_dict(source),
)
self.db.commit()
self.db.refresh(source)
return {
"id": str(source.id),
"domain": source.domain,
"name": source.name,
"created_at": source.created_at.isoformat(),
}
def _source_or_raise(self, source_id: str) -> AllowedSourceDB:
source = (
self.db.query(AllowedSourceDB)
.filter(AllowedSourceDB.id == source_id)
.first()
)
if not source:
raise NotFoundError("Source not found")
return source
def get_source(self, source_id: str) -> dict[str, Any]:
return _full_source_dict(self._source_or_raise(source_id))
def update_source(self, source_id: str, data: SourceUpdate) -> dict[str, Any]:
source = self._source_or_raise(source_id)
old_values = _source_to_dict(source)
update_data = data.model_dump(exclude_unset=True)
if "metadata" in update_data:
update_data["metadata_"] = update_data.pop("metadata")
for key, value in update_data.items():
setattr(source, key, value)
_log_audit(
self.db, "update", "source", source.id,
old_values=old_values, new_values=update_data,
)
self.db.commit()
self.db.refresh(source)
return {"status": "updated", "id": str(source.id)}
def delete_source(self, source_id: str) -> dict[str, Any]:
source = self._source_or_raise(source_id)
old_values = _source_to_dict(source)
_log_audit(self.db, "delete", "source", source.id, old_values=old_values)
self.db.query(SourceOperationDB).filter(
SourceOperationDB.source_id == source_id
).delete()
self.db.delete(source)
self.db.commit()
return {"status": "deleted", "id": source_id}
# ------------------------------------------------------------------
# Operations matrix
# ------------------------------------------------------------------
def get_operations_matrix(self) -> dict[str, Any]:
operations = self.db.query(SourceOperationDB).all()
return {
"operations": [
{
"id": str(op.id),
"source_id": str(op.source_id),
"operation": op.operation,
"allowed": op.allowed,
"conditions": op.conditions,
}
for op in operations
],
"count": len(operations),
}
def update_operation(
self, operation_id: str, data: OperationUpdate
) -> dict[str, Any]:
op = (
self.db.query(SourceOperationDB)
.filter(SourceOperationDB.id == operation_id)
.first()
)
if not op:
raise NotFoundError("Operation not found")
op.allowed = data.allowed
if data.conditions is not None:
op.conditions = data.conditions
_log_audit(
self.db, "update", "operation", op.id,
new_values={"allowed": data.allowed},
)
self.db.commit()
return {"status": "updated", "id": str(op.id)}
# ------------------------------------------------------------------
# PII rules
# ------------------------------------------------------------------
def list_pii_rules(self, category: Optional[str]) -> dict[str, Any]:
q = self.db.query(PIIRuleDB)
if category:
q = q.filter(PIIRuleDB.category == category)
rules = q.order_by(PIIRuleDB.category, PIIRuleDB.name).all()
return {
"rules": [
{
"id": str(r.id),
"name": r.name,
"description": r.description,
"pattern": r.pattern,
"category": r.category,
"action": r.action,
"active": r.active,
"created_at": r.created_at.isoformat() if r.created_at else None,
}
for r in rules
],
"count": len(rules),
}
def create_pii_rule(self, data: PIIRuleCreate) -> dict[str, Any]:
rule = PIIRuleDB(
name=data.name,
description=data.description,
pattern=data.pattern,
category=data.category,
action=data.action,
active=data.active,
)
self.db.add(rule)
_log_audit(
self.db, "create", "pii_rule", rule.id,
new_values={"name": data.name, "category": data.category},
)
self.db.commit()
self.db.refresh(rule)
return {"id": str(rule.id), "name": rule.name}
def _rule_or_raise(self, rule_id: str) -> PIIRuleDB:
rule = self.db.query(PIIRuleDB).filter(PIIRuleDB.id == rule_id).first()
if not rule:
raise NotFoundError("PII rule not found")
return rule
def update_pii_rule(self, rule_id: str, data: PIIRuleUpdate) -> dict[str, Any]:
rule = self._rule_or_raise(rule_id)
update_data = data.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(rule, key, value)
_log_audit(self.db, "update", "pii_rule", rule.id, new_values=update_data)
self.db.commit()
return {"status": "updated", "id": str(rule.id)}
def delete_pii_rule(self, rule_id: str) -> dict[str, Any]:
rule = self._rule_or_raise(rule_id)
_log_audit(
self.db, "delete", "pii_rule", rule.id,
old_values={"name": rule.name, "category": rule.category},
)
self.db.delete(rule)
self.db.commit()
return {"status": "deleted", "id": rule_id}
# ------------------------------------------------------------------
# Blocked content + audit
# ------------------------------------------------------------------
def list_blocked_content(
self,
limit: int,
offset: int,
domain: Optional[str],
date_from: Optional[str],
date_to: Optional[str],
) -> dict[str, Any]:
q = self.db.query(BlockedContentDB)
if domain:
q = q.filter(BlockedContentDB.domain == domain)
from_dt = _parse_iso_optional(date_from)
if from_dt:
q = q.filter(BlockedContentDB.created_at >= from_dt)
to_dt = _parse_iso_optional(date_to)
if to_dt:
q = q.filter(BlockedContentDB.created_at <= to_dt)
total = q.count()
entries = (
q.order_by(BlockedContentDB.created_at.desc())
.offset(offset)
.limit(limit)
.all()
)
return {
"blocked": [
{
"id": str(e.id),
"url": e.url,
"domain": e.domain,
"block_reason": e.block_reason,
"rule_id": str(e.rule_id) if e.rule_id else None,
"details": e.details,
"created_at": e.created_at.isoformat() if e.created_at else None,
}
for e in entries
],
"total": total,
}
def get_audit(
self,
limit: int,
offset: int,
entity_type: Optional[str],
date_from: Optional[str],
date_to: Optional[str],
) -> dict[str, Any]:
q = self.db.query(SourcePolicyAuditDB)
if entity_type:
q = q.filter(SourcePolicyAuditDB.entity_type == entity_type)
from_dt = _parse_iso_optional(date_from)
if from_dt:
q = q.filter(SourcePolicyAuditDB.created_at >= from_dt)
to_dt = _parse_iso_optional(date_to)
if to_dt:
q = q.filter(SourcePolicyAuditDB.created_at <= to_dt)
total = q.count()
entries = (
q.order_by(SourcePolicyAuditDB.created_at.desc())
.offset(offset)
.limit(limit)
.all()
)
return {
"entries": [
{
"id": str(e.id),
"action": e.action,
"entity_type": e.entity_type,
"entity_id": str(e.entity_id) if e.entity_id else None,
"old_values": e.old_values,
"new_values": e.new_values,
"user_id": e.user_id,
"created_at": e.created_at.isoformat() if e.created_at else None,
}
for e in entries
],
"total": total,
"limit": limit,
"offset": offset,
}
# ------------------------------------------------------------------
# Stats + report
# ------------------------------------------------------------------
def stats(self) -> dict[str, Any]:
total_sources = self.db.query(AllowedSourceDB).count()
active_sources = (
self.db.query(AllowedSourceDB).filter(AllowedSourceDB.active).count()
)
pii_rules = self.db.query(PIIRuleDB).filter(PIIRuleDB.active).count()
today_start = datetime.now(timezone.utc).replace(
hour=0, minute=0, second=0, microsecond=0
)
blocked_today = (
self.db.query(BlockedContentDB)
.filter(BlockedContentDB.created_at >= today_start)
.count()
)
blocked_total = self.db.query(BlockedContentDB).count()
return {
"active_policies": active_sources,
"allowed_sources": total_sources,
"pii_rules": pii_rules,
"blocked_today": blocked_today,
"blocked_total": blocked_total,
}
def compliance_report(self) -> dict[str, Any]:
sources = (
self.db.query(AllowedSourceDB).filter(AllowedSourceDB.active).all()
)
pii_rules = self.db.query(PIIRuleDB).filter(PIIRuleDB.active).all()
return {
"report_date": datetime.now(timezone.utc).isoformat(),
"summary": {
"active_sources": len(sources),
"active_pii_rules": len(pii_rules),
"source_types": list({s.source_type for s in sources}),
"licenses": list({s.license for s in sources if s.license}),
},
"sources": [
{
"domain": s.domain,
"name": s.name,
"license": s.license,
"legal_basis": s.legal_basis,
"trust_boost": s.trust_boost,
}
for s in sources
],
"pii_rules": [
{"name": r.name, "category": r.category, "action": r.action}
for r in pii_rules
],
}

View File

@@ -83,5 +83,7 @@ ignore_errors = False
ignore_errors = False
[mypy-compliance.api.canonical_control_routes]
ignore_errors = False
[mypy-compliance.api.source_policy_router]
ignore_errors = False
[mypy-compliance.api._http_errors]
ignore_errors = False

View File

@@ -47999,7 +47999,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response List Blocked Content Api V1 Admin Blocked Content Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48029,7 +48033,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Get Compliance Report Api V1 Admin Compliance Report Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48049,7 +48057,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Get Operations Matrix Api V1 Admin Operations Matrix Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48090,7 +48102,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Update Operation Api V1 Admin Operations Operation Id Put",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48138,7 +48154,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response List Pii Rules Api V1 Admin Pii Rules Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48176,7 +48196,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Create Pii Rule Api V1 Admin Pii Rules Post",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48217,7 +48241,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Delete Pii Rule Api V1 Admin Pii Rules Rule Id Delete",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48266,7 +48294,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Update Pii Rule Api V1 Admin Pii Rules Rule Id Put",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48369,7 +48401,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Get Policy Audit Api V1 Admin Policy Audit Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48399,7 +48435,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Get Policy Stats Api V1 Admin Policy Stats Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48463,7 +48503,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response List Sources Api V1 Admin Sources Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48501,7 +48545,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Create Source Api V1 Admin Sources Post",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48542,7 +48590,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Delete Source Api V1 Admin Sources Source Id Delete",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48581,7 +48633,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Get Source Api V1 Admin Sources Source Id Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -48630,7 +48686,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Update Source Api V1 Admin Sources Source Id Put",
"type": "object"
}
}
},
"description": "Successful Response"