feat: define governed report provider contract
This commit is contained in:
@@ -0,0 +1,311 @@
|
||||
"""Provider-neutral contracts for cross-module governed reports."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Literal, Protocol, runtime_checkable
|
||||
|
||||
|
||||
REPORT_PROVIDER_CAPABILITY_PREFIX = "reporting.report_provider."
|
||||
CAPABILITY_POLICY_REPORTING_GOVERNANCE = "policy.reporting_governance"
|
||||
CAPABILITY_REPORTING_RETENTION = "reporting.retention"
|
||||
REPORT_PROVIDER_CONTRACT_VERSION = "1.0"
|
||||
|
||||
ReportParameterType = Literal[
|
||||
"string",
|
||||
"integer",
|
||||
"number",
|
||||
"boolean",
|
||||
"date",
|
||||
"datetime",
|
||||
"reference",
|
||||
]
|
||||
ReportFieldType = Literal[
|
||||
"string",
|
||||
"integer",
|
||||
"number",
|
||||
"boolean",
|
||||
"date",
|
||||
"datetime",
|
||||
"object",
|
||||
"suppressed_count",
|
||||
]
|
||||
ReidentificationRisk = Literal["low", "moderate", "high"]
|
||||
ReportGovernanceAction = Literal["catalogue", "execute", "export"]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ReportParameterOption:
|
||||
value: str
|
||||
label: str
|
||||
description: str | None = None
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"value": self.value,
|
||||
"label": self.label,
|
||||
"description": self.description,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ReportParameterDescriptor:
|
||||
key: str
|
||||
label: str
|
||||
type: ReportParameterType = "string"
|
||||
required: bool = False
|
||||
description: str | None = None
|
||||
options_from_provider: bool = False
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"key": self.key,
|
||||
"label": self.label,
|
||||
"type": self.type,
|
||||
"required": self.required,
|
||||
"description": self.description,
|
||||
"options_from_provider": self.options_from_provider,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ReportResultField:
|
||||
path: str
|
||||
label: str
|
||||
type: ReportFieldType
|
||||
group: str
|
||||
nullable: bool = False
|
||||
sensitive: bool = False
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"path": self.path,
|
||||
"label": self.label,
|
||||
"type": self.type,
|
||||
"group": self.group,
|
||||
"nullable": self.nullable,
|
||||
"sensitive": self.sensitive,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ReportPrivacyTransform:
|
||||
id: str
|
||||
label: str
|
||||
required: bool = True
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {"id": self.id, "label": self.label, "required": self.required}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ReportDescriptor:
|
||||
provider_id: str
|
||||
report_id: str
|
||||
revision: str
|
||||
title: str
|
||||
summary: str
|
||||
parameters: tuple[ReportParameterDescriptor, ...]
|
||||
result_schema: tuple[ReportResultField, ...]
|
||||
privacy_transforms: tuple[ReportPrivacyTransform, ...]
|
||||
purpose_required: bool = True
|
||||
audience_scope_required: bool = True
|
||||
retention_class: str = "report_result"
|
||||
export_formats: tuple[str, ...] = ("json",)
|
||||
reidentification_risk: ReidentificationRisk = "moderate"
|
||||
presentation: Mapping[str, object] = field(default_factory=dict)
|
||||
contract_version: str = REPORT_PROVIDER_CONTRACT_VERSION
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"contract_version": self.contract_version,
|
||||
"provider_id": self.provider_id,
|
||||
"report_id": self.report_id,
|
||||
"revision": self.revision,
|
||||
"title": self.title,
|
||||
"summary": self.summary,
|
||||
"parameters": [item.to_dict() for item in self.parameters],
|
||||
"result_schema": [item.to_dict() for item in self.result_schema],
|
||||
"privacy_transforms": [item.to_dict() for item in self.privacy_transforms],
|
||||
"purpose_required": self.purpose_required,
|
||||
"audience_scope_required": self.audience_scope_required,
|
||||
"retention_class": self.retention_class,
|
||||
"export_formats": list(self.export_formats),
|
||||
"reidentification_risk": self.reidentification_risk,
|
||||
"presentation": dict(self.presentation),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ReportProviderRequest:
|
||||
report_id: str
|
||||
parameters: Mapping[str, object]
|
||||
purpose: str
|
||||
audience_scope: Mapping[str, object]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ReportProviderResult:
|
||||
report_id: str
|
||||
generated_at: datetime
|
||||
payload: Mapping[str, object]
|
||||
source_revisions: tuple[Mapping[str, object], ...]
|
||||
effective_scope: Mapping[str, object]
|
||||
applied_privacy_transforms: tuple[str, ...]
|
||||
provenance: Mapping[str, object]
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class ReportProvider(Protocol):
|
||||
provider_id: str
|
||||
contract_version: str
|
||||
|
||||
def list_reports(
|
||||
self,
|
||||
session: object,
|
||||
principal: object,
|
||||
) -> tuple[ReportDescriptor, ...]: ...
|
||||
|
||||
def parameter_options(
|
||||
self,
|
||||
session: object,
|
||||
principal: object,
|
||||
*,
|
||||
report_id: str,
|
||||
parameter_key: str,
|
||||
query: str,
|
||||
limit: int,
|
||||
) -> tuple[ReportParameterOption, ...]: ...
|
||||
|
||||
def execute_report(
|
||||
self,
|
||||
session: object,
|
||||
principal: object,
|
||||
*,
|
||||
request: ReportProviderRequest,
|
||||
) -> ReportProviderResult: ...
|
||||
|
||||
def authorize_result(
|
||||
self,
|
||||
session: object,
|
||||
principal: object,
|
||||
*,
|
||||
report_id: str,
|
||||
source_revisions: tuple[Mapping[str, object], ...],
|
||||
effective_scope: Mapping[str, object],
|
||||
) -> bool: ...
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ReportingGovernanceRequest:
|
||||
action: ReportGovernanceAction
|
||||
tenant_id: str
|
||||
provider_id: str
|
||||
report_id: str
|
||||
purpose: str | None
|
||||
audience_scope: Mapping[str, object]
|
||||
retention_class: str
|
||||
export_format: str | None
|
||||
reidentification_risk: ReidentificationRisk
|
||||
declared_privacy_transforms: tuple[str, ...]
|
||||
applied_privacy_transforms: tuple[str, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ReportingGovernanceDecision:
|
||||
allowed: bool
|
||||
reason: str | None
|
||||
retention_days: int | None
|
||||
export_formats: tuple[str, ...]
|
||||
required_privacy_transforms: tuple[str, ...]
|
||||
provenance: Mapping[str, object] = field(default_factory=dict)
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class ReportingGovernanceProvider(Protocol):
|
||||
def decide_reporting_action(
|
||||
self,
|
||||
session: object,
|
||||
principal: object,
|
||||
*,
|
||||
request: ReportingGovernanceRequest,
|
||||
) -> ReportingGovernanceDecision: ...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class ReportingRetentionProvider(Protocol):
|
||||
def apply_retention(
|
||||
self,
|
||||
session: object,
|
||||
*,
|
||||
dry_run: bool,
|
||||
now: datetime,
|
||||
limit: int = 500,
|
||||
) -> Mapping[str, int]: ...
|
||||
|
||||
|
||||
def report_providers(registry: object | None) -> tuple[tuple[str, ReportProvider], ...]:
|
||||
if (
|
||||
registry is None
|
||||
or not hasattr(registry, "capability_names")
|
||||
or not hasattr(registry, "capability")
|
||||
):
|
||||
return ()
|
||||
providers: list[tuple[str, ReportProvider]] = []
|
||||
for capability_name in registry.capability_names():
|
||||
if not capability_name.startswith(REPORT_PROVIDER_CAPABILITY_PREFIX):
|
||||
continue
|
||||
provider_id = capability_name.removeprefix(REPORT_PROVIDER_CAPABILITY_PREFIX)
|
||||
capability = registry.capability(capability_name)
|
||||
if not isinstance(capability, ReportProvider):
|
||||
raise TypeError(f"Invalid report provider capability: {capability_name}")
|
||||
if capability.provider_id != provider_id:
|
||||
raise ValueError(
|
||||
f"Report provider id {capability.provider_id!r} does not match "
|
||||
f"capability {capability_name!r}"
|
||||
)
|
||||
if capability.contract_version != REPORT_PROVIDER_CONTRACT_VERSION:
|
||||
raise ValueError(
|
||||
f"Unsupported report provider contract: {capability.contract_version}"
|
||||
)
|
||||
providers.append((provider_id, capability))
|
||||
return tuple(providers)
|
||||
|
||||
|
||||
def reporting_governance_provider(
|
||||
registry: object | None,
|
||||
) -> ReportingGovernanceProvider | None:
|
||||
if (
|
||||
registry is None
|
||||
or not hasattr(registry, "has_capability")
|
||||
or not registry.has_capability(CAPABILITY_POLICY_REPORTING_GOVERNANCE)
|
||||
):
|
||||
return None
|
||||
capability = registry.capability(CAPABILITY_POLICY_REPORTING_GOVERNANCE)
|
||||
if not isinstance(capability, ReportingGovernanceProvider):
|
||||
raise TypeError("Invalid reporting governance provider capability")
|
||||
return capability
|
||||
|
||||
|
||||
__all__ = [
|
||||
"CAPABILITY_POLICY_REPORTING_GOVERNANCE",
|
||||
"CAPABILITY_REPORTING_RETENTION",
|
||||
"REPORT_PROVIDER_CAPABILITY_PREFIX",
|
||||
"REPORT_PROVIDER_CONTRACT_VERSION",
|
||||
"ReportDescriptor",
|
||||
"ReportParameterDescriptor",
|
||||
"ReportParameterOption",
|
||||
"ReportPrivacyTransform",
|
||||
"ReportProvider",
|
||||
"ReportProviderRequest",
|
||||
"ReportProviderResult",
|
||||
"ReportResultField",
|
||||
"ReportingGovernanceDecision",
|
||||
"ReportingGovernanceProvider",
|
||||
"ReportingGovernanceRequest",
|
||||
"ReportingRetentionProvider",
|
||||
"report_providers",
|
||||
"reporting_governance_provider",
|
||||
]
|
||||
Reference in New Issue
Block a user