Define governed View projection contract
This commit is contained in:
@@ -18,12 +18,27 @@ DefinitionGovernanceAction = Literal[
|
||||
"derive",
|
||||
"automate",
|
||||
]
|
||||
ViewGovernanceAction = Literal[
|
||||
"view",
|
||||
"select",
|
||||
"assign",
|
||||
"edit",
|
||||
"derive",
|
||||
"workflow_activate",
|
||||
]
|
||||
|
||||
CAPABILITY_POLICY_PRIVACY_RETENTION = "policy.privacyRetention"
|
||||
CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY = "policy.schedulingParticipantPrivacy"
|
||||
CAPABILITY_POLICY_DEFINITION_GOVERNANCE = "policy.definitionGovernance"
|
||||
CAPABILITY_POLICY_VIEW_GOVERNANCE = "policy.viewGovernance"
|
||||
|
||||
POLICY_SCOPE_TYPES: tuple[PolicyScopeType, ...] = ("system", "tenant", "user", "group", "campaign")
|
||||
POLICY_SCOPE_TYPES: tuple[PolicyScopeType, ...] = (
|
||||
"system",
|
||||
"tenant",
|
||||
"user",
|
||||
"group",
|
||||
"campaign",
|
||||
)
|
||||
|
||||
|
||||
def normalize_policy_scope_type(scope_type: str) -> PolicyScopeType:
|
||||
@@ -43,7 +58,11 @@ class PolicySourceRef:
|
||||
return policy_source_path(self.scope_type, self.scope_id)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {"scope_type": self.scope_type, "scope_id": self.scope_id, "path": self.path}
|
||||
return {
|
||||
"scope_type": self.scope_type,
|
||||
"scope_id": self.scope_id,
|
||||
"path": self.path,
|
||||
}
|
||||
|
||||
|
||||
def policy_source_path(scope_type: str, scope_id: str | None = None) -> str:
|
||||
@@ -53,7 +72,9 @@ def policy_source_path(scope_type: str, scope_id: str | None = None) -> str:
|
||||
raise ValueError("System policy sources do not carry a scope_id")
|
||||
return "system"
|
||||
if not scope_id:
|
||||
raise ValueError(f"{clean_scope.capitalize()} policy sources require a scope_id")
|
||||
raise ValueError(
|
||||
f"{clean_scope.capitalize()} policy sources require a scope_id"
|
||||
)
|
||||
return f"{clean_scope}:{quote(str(scope_id), safe='')}"
|
||||
|
||||
|
||||
@@ -63,7 +84,9 @@ def parse_policy_source_path(path: str) -> PolicySourceRef:
|
||||
return PolicySourceRef(scope_type="system")
|
||||
scope_type, separator, encoded_scope_id = clean_path.partition(":")
|
||||
if not separator:
|
||||
raise ValueError("Policy source path must be system or <scope_type>:<url-encoded-scope-id>")
|
||||
raise ValueError(
|
||||
"Policy source path must be system or <scope_type>:<url-encoded-scope-id>"
|
||||
)
|
||||
clean_scope = normalize_policy_scope_type(scope_type)
|
||||
if clean_scope == "system":
|
||||
raise ValueError("System policy source path must be exactly system")
|
||||
@@ -100,9 +123,13 @@ class PolicySourceStep:
|
||||
policy_value = value.get("policy")
|
||||
return cls(
|
||||
scope_type=normalize_policy_scope_type(str(value.get("scope_type", ""))),
|
||||
scope_id=str(value["scope_id"]) if value.get("scope_id") is not None else None,
|
||||
scope_id=str(value["scope_id"])
|
||||
if value.get("scope_id") is not None
|
||||
else None,
|
||||
label=str(value.get("label") or ""),
|
||||
applied_fields=tuple(str(field) for field in (value.get("applied_fields") or ())),
|
||||
applied_fields=tuple(
|
||||
str(field) for field in (value.get("applied_fields") or ())
|
||||
),
|
||||
policy=policy_value if isinstance(policy_value, Mapping) else {},
|
||||
)
|
||||
|
||||
@@ -176,8 +203,7 @@ class DefinitionGovernancePolicy(Protocol):
|
||||
session: object | None = None,
|
||||
*,
|
||||
request: DefinitionGovernanceRequest,
|
||||
) -> PolicyDecision:
|
||||
...
|
||||
) -> PolicyDecision: ...
|
||||
|
||||
|
||||
def definition_governance_policy(
|
||||
@@ -186,17 +212,79 @@ def definition_governance_policy(
|
||||
if (
|
||||
registry is None
|
||||
or not hasattr(registry, "has_capability")
|
||||
or not registry.has_capability(
|
||||
CAPABILITY_POLICY_DEFINITION_GOVERNANCE
|
||||
)
|
||||
or not registry.has_capability(CAPABILITY_POLICY_DEFINITION_GOVERNANCE)
|
||||
):
|
||||
return None
|
||||
capability = registry.capability(CAPABILITY_POLICY_DEFINITION_GOVERNANCE)
|
||||
return (
|
||||
capability
|
||||
if isinstance(capability, DefinitionGovernancePolicy)
|
||||
else None
|
||||
)
|
||||
return capability if isinstance(capability, DefinitionGovernancePolicy) else None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ViewGovernanceRequest:
|
||||
tenant_id: str
|
||||
action: ViewGovernanceAction
|
||||
actor: PrincipalRef
|
||||
target_scope: DefinitionScopeRef
|
||||
view_id: str | None = None
|
||||
candidate_view_ids: tuple[str, ...] = ()
|
||||
candidate_surface_ids: tuple[str, ...] = ()
|
||||
requested_surface_ids: tuple[str, ...] = ()
|
||||
context: Mapping[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ViewGovernanceDecision:
|
||||
allowed: bool
|
||||
reason: str | None = None
|
||||
allowed_view_ids: frozenset[str] | None = None
|
||||
visible_surface_ids: frozenset[str] | None = None
|
||||
source_path: tuple[PolicySourceStep, ...] = ()
|
||||
requirements: tuple[str, ...] = ()
|
||||
diagnostics: tuple[Mapping[str, Any], ...] = ()
|
||||
details: Mapping[str, Any] = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"allowed": self.allowed,
|
||||
"reason": self.reason,
|
||||
"allowed_view_ids": (
|
||||
sorted(self.allowed_view_ids)
|
||||
if self.allowed_view_ids is not None
|
||||
else None
|
||||
),
|
||||
"visible_surface_ids": (
|
||||
sorted(self.visible_surface_ids)
|
||||
if self.visible_surface_ids is not None
|
||||
else None
|
||||
),
|
||||
"source_path": [step.to_dict() for step in self.source_path],
|
||||
"requirements": list(self.requirements),
|
||||
"diagnostics": [dict(item) for item in self.diagnostics],
|
||||
"details": dict(self.details),
|
||||
}
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class ViewGovernancePolicy(Protocol):
|
||||
def resolve_view_action(
|
||||
self,
|
||||
session: object | None = None,
|
||||
*,
|
||||
request: ViewGovernanceRequest,
|
||||
) -> ViewGovernanceDecision: ...
|
||||
|
||||
|
||||
def view_governance_policy(
|
||||
registry: object | None,
|
||||
) -> ViewGovernancePolicy | None:
|
||||
if (
|
||||
registry is None
|
||||
or not hasattr(registry, "has_capability")
|
||||
or not registry.has_capability(CAPABILITY_POLICY_VIEW_GOVERNANCE)
|
||||
):
|
||||
return None
|
||||
capability = registry.capability(CAPABILITY_POLICY_VIEW_GOVERNANCE)
|
||||
return capability if isinstance(capability, ViewGovernancePolicy) else None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
@@ -244,44 +332,32 @@ class SchedulingParticipantPrivacyPolicy(Protocol):
|
||||
session: object,
|
||||
*,
|
||||
request: SchedulingParticipantPrivacyRequest,
|
||||
) -> SchedulingParticipantPrivacyDecision:
|
||||
...
|
||||
) -> SchedulingParticipantPrivacyDecision: ...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class PrivacyRetentionService(Protocol):
|
||||
def privacy_policy_from_settings(self, *args: Any, **kwargs: Any) -> Any:
|
||||
...
|
||||
def privacy_policy_from_settings(self, *args: Any, **kwargs: Any) -> Any: ...
|
||||
|
||||
def privacy_policy_from_session(self, *args: Any, **kwargs: Any) -> Any:
|
||||
...
|
||||
def privacy_policy_from_session(self, *args: Any, **kwargs: Any) -> Any: ...
|
||||
|
||||
def set_privacy_policy(self, *args: Any, **kwargs: Any) -> Any:
|
||||
...
|
||||
def set_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: ...
|
||||
|
||||
def effective_privacy_policy(self, *args: Any, **kwargs: Any) -> Any:
|
||||
...
|
||||
def effective_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: ...
|
||||
|
||||
def parent_privacy_policy(self, *args: Any, **kwargs: Any) -> Any:
|
||||
...
|
||||
def parent_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: ...
|
||||
|
||||
def effective_privacy_policy_sources(self, *args: Any, **kwargs: Any) -> Any:
|
||||
...
|
||||
def effective_privacy_policy_sources(self, *args: Any, **kwargs: Any) -> Any: ...
|
||||
|
||||
def parent_privacy_policy_sources(self, *args: Any, **kwargs: Any) -> Any:
|
||||
...
|
||||
def parent_privacy_policy_sources(self, *args: Any, **kwargs: Any) -> Any: ...
|
||||
|
||||
def get_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any:
|
||||
...
|
||||
def get_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any: ...
|
||||
|
||||
def set_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any:
|
||||
...
|
||||
def set_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any: ...
|
||||
|
||||
def sanitize_audit_details_for_policy(self, *args: Any, **kwargs: Any) -> Any:
|
||||
...
|
||||
def sanitize_audit_details_for_policy(self, *args: Any, **kwargs: Any) -> Any: ...
|
||||
|
||||
def apply_retention_policy(self, *args: Any, **kwargs: Any) -> Any:
|
||||
...
|
||||
def apply_retention_policy(self, *args: Any, **kwargs: Any) -> Any: ...
|
||||
|
||||
|
||||
def scheduling_participant_privacy_policy(
|
||||
@@ -294,4 +370,8 @@ def scheduling_participant_privacy_policy(
|
||||
if not registry.has_capability(CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY):
|
||||
return None
|
||||
capability = registry.capability(CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY)
|
||||
return capability if isinstance(capability, SchedulingParticipantPrivacyPolicy) else None
|
||||
return (
|
||||
capability
|
||||
if isinstance(capability, SchedulingParticipantPrivacyPolicy)
|
||||
else None
|
||||
)
|
||||
|
||||
@@ -43,6 +43,7 @@ class EffectiveView:
|
||||
name: str | None
|
||||
visible_surface_ids: frozenset[str]
|
||||
locked: bool = False
|
||||
projection_active: bool = False
|
||||
provenance: tuple[dict[str, object], ...] = ()
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user