feat(core): define datasource visibility policy contract

src/govoplan_core/core/datasources.py

@@ -5,6 +5,7 @@ from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Literal, Protocol, runtime_checkable
 
+from govoplan_core.core.access import PrincipalRef
 from govoplan_core.core.external_references import (
     SOURCE_AUTHORITY_MODES,
     SourceAuthorityMode,
@@ -24,6 +25,7 @@ CAPABILITY_DATASOURCE_LIFECYCLE = "datasources.lifecycle"
 CAPABILITY_DATASOURCE_PUBLICATION = "datasources.publication"
 CAPABILITY_DATASOURCE_ORIGINS = "connectors.datasourceOrigins"
 CAPABILITY_DATASOURCE_ARTIFACT_BACKENDS = "datasources.artifactBackends"
+CAPABILITY_POLICY_DATASOURCE_VISIBILITY = "policy.datasourceVisibility"
 
 DatasourceMode = Literal["live", "cached", "static"]
 DatasourceKind = Literal[
@@ -38,6 +40,7 @@ DatasourceKind = Literal[
 ]
 DatasourceShape = Literal["tabular", "document", "binary", "directory", "stream"]
 DatasourceConsistency = Literal["current", "live", "frozen"]
+DatasourceVisibilityAction = Literal["discover", "read"]
 DatasourcePublicationStatus = Literal[
     "published",
     "published_with_warnings",
@@ -70,6 +73,7 @@ class DatasourceField:
     name: str
     data_type: str
     nullable: bool = True
+    classification: str = "internal"
 
 
 @dataclass(frozen=True, slots=True)
@@ -90,6 +94,8 @@ class DatasourceGovernance:
     classification: str = "internal"
     privacy_profile_ref: str | None = None
     retention_policy_ref: str | None = None
+    access_policy_ref: str | None = None
+    visibility_policy: Mapping[str, object] = field(default_factory=dict)
     hold_refs: tuple[str, ...] = ()
     publication_state: str = "draft"
     transfer_agreement_ref: str | None = None
@@ -160,6 +166,12 @@ class DatasourceGovernance:
             retention_policy_ref=_optional_governance_text(
                 source.get("retention_policy_ref")
             ),
+            access_policy_ref=_optional_governance_text(
+                source.get("access_policy_ref")
+            ),
+            visibility_policy=_governance_mapping(
+                source.get("visibility_policy")
+            ),
             hold_refs=_governance_texts(source.get("hold_refs")),
             publication_state=str(source.get("publication_state") or "draft"),
             transfer_agreement_ref=_optional_governance_text(
@@ -191,6 +203,8 @@ class DatasourceGovernance:
             "classification": self.classification,
             "privacy_profile_ref": self.privacy_profile_ref,
             "retention_policy_ref": self.retention_policy_ref,
+            "access_policy_ref": self.access_policy_ref,
+            "visibility_policy": dict(self.visibility_policy),
             "hold_refs": list(self.hold_refs),
             "publication_state": self.publication_state,
             "transfer_agreement_ref": self.transfer_agreement_ref,
@@ -203,6 +217,39 @@ class DatasourceGovernance:
         }
 
 
+@dataclass(frozen=True, slots=True)
+class DatasourceVisibilityPolicyRequest:
+    tenant_id: str
+    datasource_ref: str
+    principal: PrincipalRef
+    action: DatasourceVisibilityAction
+    classification: str = "internal"
+    policy_ref: str | None = None
+    consistency: DatasourceConsistency = "current"
+    materialization_ref: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class DatasourceVisibilityPolicyDecision:
+    allowed: bool
+    reason: str | None = None
+    policies: tuple[Mapping[str, object], ...] = ()
+    decision_ref: str | None = None
+    provenance: Mapping[str, object] = field(default_factory=dict)
+
+
+@runtime_checkable
+class DatasourceVisibilityPolicyProvider(Protocol):
+    """Optionally tighten Datasources-owned local visibility policy."""
+
+    def decide_datasource_visibility(
+        self,
+        session: object,
+        *,
+        request: DatasourceVisibilityPolicyRequest,
+    ) -> DatasourceVisibilityPolicyDecision: ...
+
+
 @dataclass(frozen=True, slots=True)
 class DatasourceDescriptor:
     ref: str
@@ -659,6 +706,13 @@ def datasource_origins(registry: object | None) -> DatasourceOriginProvider | No
     return capability if isinstance(capability, DatasourceOriginProvider) else None
 
 
+def datasource_visibility_policy_provider(
+    registry: object | None,
+) -> DatasourceVisibilityPolicyProvider | None:
+    capability = _capability(registry, CAPABILITY_POLICY_DATASOURCE_VISIBILITY)
+    return capability if isinstance(capability, DatasourceVisibilityPolicyProvider) else None
+
+
 def _capability(registry: object | None, name: str) -> object | None:
     if (
         registry is None
@@ -695,6 +749,7 @@ __all__ = [
     "CAPABILITY_DATASOURCE_LIFECYCLE",
     "CAPABILITY_DATASOURCE_ORIGINS",
     "CAPABILITY_DATASOURCE_PUBLICATION",
+    "CAPABILITY_POLICY_DATASOURCE_VISIBILITY",
     "DatasourceAccessError",
     "DatasourceArtifactReference",
     "DatasourceArtifactBackend",
@@ -725,9 +780,14 @@ __all__ = [
     "DatasourceStageInput",
     "DatasourceUnavailableError",
     "DatasourceValidationError",
+    "DatasourceVisibilityAction",
+    "DatasourceVisibilityPolicyDecision",
+    "DatasourceVisibilityPolicyProvider",
+    "DatasourceVisibilityPolicyRequest",
     "datasource_catalogue",
     "datasource_artifact_backend_provider",
     "datasource_lifecycle",
     "datasource_origins",
     "datasource_publication",
+    "datasource_visibility_policy_provider",
 ]