feat: implement governed datasource catalogue metadata

2026-08-01 17:48:28 +02:00
parent 8da8035693
commit 98d93e707f
14 changed files with 1382 additions and 12 deletions
@@ -30,6 +30,21 @@ class DatasourceRecord(Base, TimestampMixin):
        Index("ix_datasource_catalogue_tenant_status", "tenant_id", "status"),
        Index("ix_datasource_catalogue_tenant_mode", "tenant_id", "mode"),
        Index("ix_datasource_catalogue_origin", "tenant_id", "provider", "provider_ref"),
+        Index(
+            "ix_datasource_catalogue_tenant_authority",
+            "tenant_id",
+            "authority_mode",
+        ),
+        Index(
+            "ix_datasource_catalogue_tenant_classification",
+            "tenant_id",
+            "classification",
+        ),
+        Index(
+            "ix_datasource_catalogue_tenant_publication",
+            "tenant_id",
+            "publication_state",
+        ),
    )

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
@@ -70,6 +85,49 @@ class DatasourceRecord(Base, TimestampMixin):
        default=dict,
        nullable=False,
    )
+    owner_ref: Mapped[str | None] = mapped_column(String(500), nullable=True, index=True)
+    steward_ref: Mapped[str | None] = mapped_column(String(500), nullable=True, index=True)
+    responsible_organization_ref: Mapped[str | None] = mapped_column(
+        String(500), nullable=True, index=True
+    )
+    responsible_function_ref: Mapped[str | None] = mapped_column(
+        String(500), nullable=True, index=True
+    )
+    authoritative_source_ref: Mapped[str | None] = mapped_column(
+        String(1000), nullable=True
+    )
+    authority_mode: Mapped[str] = mapped_column(
+        String(40), default="linked_reference", nullable=False, index=True
+    )
+    legal_basis_refs: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
+    purposes: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
+    semantic_definition: Mapped[str | None] = mapped_column(Text, nullable=True)
+    schema_owner_ref: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    official_keys: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
+    classification: Mapped[str] = mapped_column(
+        String(80), default="internal", nullable=False, index=True
+    )
+    privacy_profile_ref: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    retention_policy_ref: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    hold_refs: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
+    publication_state: Mapped[str] = mapped_column(
+        String(50), default="draft", nullable=False, index=True
+    )
+    transfer_agreement_ref: Mapped[str | None] = mapped_column(
+        String(500), nullable=True
+    )
+    freshness_policy: Mapped[dict[str, Any]] = mapped_column(
+        JSON, default=dict, nullable=False
+    )
+    quality_policy: Mapped[dict[str, Any]] = mapped_column(
+        JSON, default=dict, nullable=False
+    )
+    known_limits: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
+    correction_procedure_ref: Mapped[str | None] = mapped_column(
+        String(500), nullable=True
+    )
+    affected_refs: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
+    dependency_refs: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
    created_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
    updated_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
    deleted_at: Mapped[datetime | None] = mapped_column(
@@ -83,6 +141,49 @@ class DatasourceRecord(Base, TimestampMixin):
        cascade="all, delete-orphan",
        order_by="DatasourceMaterializationRecord.revision",
    )
+    governance_references: Mapped[
+        list["DatasourceGovernanceReferenceRecord"]
+    ] = relationship(
+        back_populates="datasource",
+        cascade="all, delete-orphan",
+    )
+
+
+class DatasourceGovernanceReferenceRecord(Base):
+    __tablename__ = "datasource_governance_references"
+    __table_args__ = (
+        UniqueConstraint(
+            "datasource_id",
+            "relation",
+            "reference",
+            name="uq_datasource_governance_reference",
+        ),
+        Index(
+            "ix_datasource_governance_reference_lookup",
+            "tenant_id",
+            "relation",
+            "reference",
+        ),
+        Index(
+            "ix_datasource_governance_reference_source",
+            "tenant_id",
+            "datasource_id",
+            "relation",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    datasource_id: Mapped[str] = mapped_column(
+        ForeignKey("datasource_catalogue.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    relation: Mapped[str] = mapped_column(String(30), nullable=False)
+    reference: Mapped[str] = mapped_column(String(1000), nullable=False)
+
+    datasource: Mapped[DatasourceRecord] = relationship(
+        back_populates="governance_references"
+    )


 class DatasourceMaterializationRecord(Base, TimestampMixin):
@@ -155,6 +256,12 @@ class DatasourceMaterializationRecord(Base, TimestampMixin):
        default=dict,
        nullable=False,
    )
+    governance_snapshot_: Mapped[dict[str, Any]] = mapped_column(
+        "governance_snapshot",
+        JSON,
+        default=dict,
+        nullable=False,
+    )
    created_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)

    datasource: Mapped[DatasourceRecord] = relationship(back_populates="materializations")
@@ -300,6 +407,12 @@ class DatasourceStageRecord(Base, TimestampMixin):
        default=dict,
        nullable=False,
    )
+    governance_: Mapped[dict[str, Any]] = mapped_column(
+        "governance",
+        JSON,
+        default=dict,
+        nullable=False,
+    )
    promoted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    promoted_materialization_id: Mapped[str | None] = mapped_column(
        String(36),
@@ -29,6 +29,11 @@ from govoplan_core.core.modules import (
    PermissionDefinition,
    RoleTemplate,
 )
+from govoplan_core.core.provider_governance import (
+    ModuleArchitectureDeclaration,
+    ModuleArchitectureDocumentation,
+    ModuleMaturityEvidence,
+)
 from govoplan_core.db.base import Base
 from govoplan_datasources.backend.db import models as datasource_models
 from govoplan_datasources.backend.service import (
@@ -45,6 +50,58 @@ MODULE_NAME = "Datasources"
 MODULE_VERSION = "0.1.14"
 DATASOURCE_INTERFACE_VERSION = "0.1.0"

+ARCHITECTURE = ModuleArchitectureDeclaration(
+    layer="data_reporting_integration",
+    kind="foundation",
+    maturity="vertical_slice",
+    evidence=(
+        ModuleMaturityEvidence(
+            kind="test",
+            reference="tests/test_lifecycle.py",
+            summary="Exercises staging, immutable materialization, publication, and typed governance behavior.",
+        ),
+        ModuleMaturityEvidence(
+            kind="documentation",
+            reference="docs/CONCEPT.md",
+            summary="Defines datasource ownership and connector/dataflow boundaries.",
+        ),
+        ModuleMaturityEvidence(
+            kind="migration",
+            reference="tests/test_migrations.py",
+            summary="Migrates and validates the governed catalogue schema.",
+        ),
+    ),
+    known_limits=(
+        "Governance references are stable provider-neutral refs; dedicated selectors depend on the owning optional modules.",
+        "Quality and freshness policies are stored and snapshotted but enforcement remains provider-specific.",
+    ),
+    supported_authority_modes=(
+        "native_authoritative",
+        "external_authoritative",
+        "external_mirror",
+        "governed_sync",
+        "governance_overlay",
+        "linked_reference",
+    ),
+    owned_concepts=(
+        "datasource catalogue identity",
+        "datasource governance metadata",
+        "staging and immutable materializations",
+        "datasource publication lifecycle",
+    ),
+    non_owned_concepts=(
+        "external transport and credentials",
+        "dataflow transformation semantics",
+        "report presentation",
+    ),
+    documentation=ModuleArchitectureDocumentation(
+        migration=("tests/test_migrations.py",),
+        recovery=("docs/CONCEPT.md",),
+        security=("docs/CONCEPT.md",),
+        operations=("docs/CONCEPT.md",),
+    ),
+)
+

 def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
    module_id, resource, action = scope.split(":", 2)
@@ -265,18 +322,23 @@ manifest = ModuleManifest(
            label="Datasources",
        ),
    ),
+    architecture=ARCHITECTURE,
    documentation=(
        DocumentationTopic(
            id="datasources.lifecycle",
            title="Datasource lifecycle",
-            summary="Governed live, cached, and static data with staging and frozen states.",
+            summary="Governed live, cached, and static data/register entries with staging, frozen states, provenance, and institutional ownership context.",
            body=(
                "Datasources is the provider-neutral catalogue consumed by Dataflow, "
                "Workflow, Reporting, and policy-aware modules. Static data is staged "
                "before promotion. Cached data refreshes connector origins into immutable "
                "materializations. Live data is read through a connector and may be frozen "
                "for reproducible evidence. Connectors owns protocols and credentials; "
-                "Datasources owns data identity, provenance, lifecycle, and read semantics."
+                "Datasources owns data identity, provenance, lifecycle, read semantics, "
+                "and the typed governance catalogue for authority, purpose, quality, "
+                "freshness, classification, correction, and dependent services, flows, "
+                "reports, controls, and decisions. Governance metadata visibility does not "
+                "grant access to protected rows."
            ),
            layer="available",
            documentation_types=("admin", "user"),
@@ -0,0 +1,216 @@
+"""v0.1.14 governed datasource catalogue
+
+Revision ID: a7c1e4d9b2f6
+Revises: d5f0a2b8c3e7
+Create Date: 2026-08-01 00:00:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "a7c1e4d9b2f6"
+down_revision = "d5f0a2b8c3e7"
+branch_labels = None
+depends_on = None
+
+
+_JSON_LIST = sa.text("'[]'")
+_JSON_OBJECT = sa.text("'{}'")
+
+
+def upgrade() -> None:
+    for column in (
+        sa.Column("owner_ref", sa.String(length=500), nullable=True),
+        sa.Column("steward_ref", sa.String(length=500), nullable=True),
+        sa.Column("responsible_organization_ref", sa.String(length=500), nullable=True),
+        sa.Column("responsible_function_ref", sa.String(length=500), nullable=True),
+        sa.Column("authoritative_source_ref", sa.String(length=1000), nullable=True),
+        sa.Column("authority_mode", sa.String(length=40), nullable=False, server_default="linked_reference"),
+        sa.Column("legal_basis_refs", sa.JSON(), nullable=False, server_default=_JSON_LIST),
+        sa.Column("purposes", sa.JSON(), nullable=False, server_default=_JSON_LIST),
+        sa.Column("semantic_definition", sa.Text(), nullable=True),
+        sa.Column("schema_owner_ref", sa.String(length=500), nullable=True),
+        sa.Column("official_keys", sa.JSON(), nullable=False, server_default=_JSON_LIST),
+        sa.Column("classification", sa.String(length=80), nullable=False, server_default="internal"),
+        sa.Column("privacy_profile_ref", sa.String(length=500), nullable=True),
+        sa.Column("retention_policy_ref", sa.String(length=500), nullable=True),
+        sa.Column("hold_refs", sa.JSON(), nullable=False, server_default=_JSON_LIST),
+        sa.Column("publication_state", sa.String(length=50), nullable=False, server_default="internal"),
+        sa.Column("transfer_agreement_ref", sa.String(length=500), nullable=True),
+        sa.Column("freshness_policy", sa.JSON(), nullable=False, server_default=_JSON_OBJECT),
+        sa.Column("quality_policy", sa.JSON(), nullable=False, server_default=_JSON_OBJECT),
+        sa.Column("known_limits", sa.JSON(), nullable=False, server_default=_JSON_LIST),
+        sa.Column("correction_procedure_ref", sa.String(length=500), nullable=True),
+        sa.Column("affected_refs", sa.JSON(), nullable=False, server_default=_JSON_LIST),
+        sa.Column("dependency_refs", sa.JSON(), nullable=False, server_default=_JSON_LIST),
+    ):
+        op.add_column("datasource_catalogue", column)
+
+    op.add_column(
+        "datasource_materializations",
+        sa.Column(
+            "governance_snapshot",
+            sa.JSON(),
+            nullable=False,
+            server_default=_JSON_OBJECT,
+        ),
+    )
+    op.add_column(
+        "datasource_stages",
+        sa.Column("governance", sa.JSON(), nullable=False, server_default=_JSON_OBJECT),
+    )
+
+    _backfill_governance()
+
+    for name, columns in (
+        ("ix_datasource_catalogue_owner_ref", ["owner_ref"]),
+        ("ix_datasource_catalogue_steward_ref", ["steward_ref"]),
+        ("ix_datasource_catalogue_responsible_organization_ref", ["responsible_organization_ref"]),
+        ("ix_datasource_catalogue_responsible_function_ref", ["responsible_function_ref"]),
+        ("ix_datasource_catalogue_authority_mode", ["authority_mode"]),
+        ("ix_datasource_catalogue_classification", ["classification"]),
+        ("ix_datasource_catalogue_publication_state", ["publication_state"]),
+        ("ix_datasource_catalogue_tenant_authority", ["tenant_id", "authority_mode"]),
+        ("ix_datasource_catalogue_tenant_classification", ["tenant_id", "classification"]),
+        ("ix_datasource_catalogue_tenant_publication", ["tenant_id", "publication_state"]),
+    ):
+        op.create_index(name, "datasource_catalogue", columns, unique=False)
+
+
+def downgrade() -> None:
+    for name in (
+        "ix_datasource_catalogue_tenant_publication",
+        "ix_datasource_catalogue_tenant_classification",
+        "ix_datasource_catalogue_tenant_authority",
+        "ix_datasource_catalogue_publication_state",
+        "ix_datasource_catalogue_classification",
+        "ix_datasource_catalogue_authority_mode",
+        "ix_datasource_catalogue_responsible_function_ref",
+        "ix_datasource_catalogue_responsible_organization_ref",
+        "ix_datasource_catalogue_steward_ref",
+        "ix_datasource_catalogue_owner_ref",
+    ):
+        op.drop_index(name, table_name="datasource_catalogue")
+    op.drop_column("datasource_stages", "governance")
+    op.drop_column("datasource_materializations", "governance_snapshot")
+    for name in (
+        "dependency_refs",
+        "affected_refs",
+        "correction_procedure_ref",
+        "known_limits",
+        "quality_policy",
+        "freshness_policy",
+        "transfer_agreement_ref",
+        "publication_state",
+        "hold_refs",
+        "retention_policy_ref",
+        "privacy_profile_ref",
+        "classification",
+        "official_keys",
+        "schema_owner_ref",
+        "semantic_definition",
+        "purposes",
+        "legal_basis_refs",
+        "authority_mode",
+        "authoritative_source_ref",
+        "responsible_function_ref",
+        "responsible_organization_ref",
+        "steward_ref",
+        "owner_ref",
+    ):
+        op.drop_column("datasource_catalogue", name)
+
+
+def _backfill_governance() -> None:
+    connection = op.get_bind()
+    catalogue = sa.table(
+        "datasource_catalogue",
+        sa.column("id", sa.String()),
+        sa.column("mode", sa.String()),
+        sa.column("provider_ref", sa.String()),
+        sa.column("authority_mode", sa.String()),
+        sa.column("authoritative_source_ref", sa.String()),
+        sa.column("purposes", sa.JSON()),
+        sa.column("publication_state", sa.String()),
+    )
+    materializations = sa.table(
+        "datasource_materializations",
+        sa.column("id", sa.String()),
+        sa.column("datasource_id", sa.String()),
+        sa.column("governance_snapshot", sa.JSON()),
+    )
+    stages = sa.table(
+        "datasource_stages",
+        sa.column("id", sa.String()),
+        sa.column("mode", sa.String()),
+        sa.column("provider_ref", sa.String()),
+        sa.column("governance", sa.JSON()),
+    )
+
+    snapshots: dict[str, dict[str, object]] = {}
+    for row in connection.execute(
+        sa.select(catalogue.c.id, catalogue.c.mode, catalogue.c.provider_ref)
+    ).mappings():
+        provider_ref = str(row["provider_ref"] or "").strip() or None
+        mode = str(row["mode"] or "")
+        authority_mode = (
+            "external_authoritative"
+            if mode == "live"
+            else "external_mirror"
+            if mode == "cached" and provider_ref
+            else "native_authoritative"
+        )
+        snapshot = {
+            "authority_mode": authority_mode,
+            "authoritative_source_ref": provider_ref,
+            "classification": "internal",
+            "publication_state": "internal",
+            "purposes": ["governed_data_processing"],
+        }
+        snapshots[str(row["id"])] = snapshot
+        connection.execute(
+            sa.update(catalogue)
+            .where(catalogue.c.id == row["id"])
+            .values(
+                authority_mode=authority_mode,
+                authoritative_source_ref=provider_ref,
+                purposes=["governed_data_processing"],
+                publication_state="internal",
+            )
+        )
+
+    for row in connection.execute(
+        sa.select(materializations.c.id, materializations.c.datasource_id)
+    ).mappings():
+        connection.execute(
+            sa.update(materializations)
+            .where(materializations.c.id == row["id"])
+            .values(
+                governance_snapshot=snapshots.get(str(row["datasource_id"]), {})
+            )
+        )
+
+    for row in connection.execute(
+        sa.select(stages.c.id, stages.c.mode, stages.c.provider_ref)
+    ).mappings():
+        provider_ref = str(row["provider_ref"] or "").strip() or None
+        mode = str(row["mode"] or "")
+        connection.execute(
+            sa.update(stages)
+            .where(stages.c.id == row["id"])
+            .values(
+                governance={
+                    "authority_mode": (
+                        "external_mirror"
+                        if mode == "cached" and provider_ref
+                        else "native_authoritative"
+                    ),
+                    "authoritative_source_ref": provider_ref,
+                    "classification": "internal",
+                    "publication_state": "internal",
+                    "purposes": ["governed_data_processing"],
+                }
+            )
+        )
@@ -0,0 +1,121 @@
+"""v0.1.14 datasource governance reference index
+
+Revision ID: b8d2f5a0c3e7
+Revises: a7c1e4d9b2f6
+Create Date: 2026-08-01 00:00:01.000000
+"""
+from __future__ import annotations
+
+import uuid
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "b8d2f5a0c3e7"
+down_revision = "a7c1e4d9b2f6"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "datasource_governance_references",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("datasource_id", sa.String(length=36), nullable=False),
+        sa.Column("relation", sa.String(length=30), nullable=False),
+        sa.Column("reference", sa.String(length=1000), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["datasource_id"],
+            ["datasource_catalogue.id"],
+            name=op.f(
+                "fk_datasource_governance_references_datasource_id_"
+                "datasource_catalogue"
+            ),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint(
+            "id", name=op.f("pk_datasource_governance_references")
+        ),
+        sa.UniqueConstraint(
+            "datasource_id",
+            "relation",
+            "reference",
+            name="uq_datasource_governance_reference",
+        ),
+    )
+    op.create_index(
+        op.f("ix_datasource_governance_references_tenant_id"),
+        "datasource_governance_references",
+        ["tenant_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_datasource_governance_reference_lookup",
+        "datasource_governance_references",
+        ["tenant_id", "relation", "reference"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_datasource_governance_reference_source",
+        "datasource_governance_references",
+        ["tenant_id", "datasource_id", "relation"],
+        unique=False,
+    )
+    _backfill_references()
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_datasource_governance_reference_source",
+        table_name="datasource_governance_references",
+    )
+    op.drop_index(
+        "ix_datasource_governance_reference_lookup",
+        table_name="datasource_governance_references",
+    )
+    op.drop_index(
+        op.f("ix_datasource_governance_references_tenant_id"),
+        table_name="datasource_governance_references",
+    )
+    op.drop_table("datasource_governance_references")
+
+
+def _backfill_references() -> None:
+    connection = op.get_bind()
+    catalogue = sa.table(
+        "datasource_catalogue",
+        sa.column("id", sa.String()),
+        sa.column("tenant_id", sa.String()),
+        sa.column("affected_refs", sa.JSON()),
+        sa.column("dependency_refs", sa.JSON()),
+    )
+    references = sa.table(
+        "datasource_governance_references",
+        sa.column("id", sa.String()),
+        sa.column("tenant_id", sa.String()),
+        sa.column("datasource_id", sa.String()),
+        sa.column("relation", sa.String()),
+        sa.column("reference", sa.String()),
+    )
+    for row in connection.execute(sa.select(catalogue)).mappings():
+        for relation, values in (
+            ("affected", row["affected_refs"]),
+            ("depends_on", row["dependency_refs"]),
+        ):
+            normalized = {
+                str(value).strip()
+                for value in values or ()
+                if str(value).strip()
+            }
+            for reference in sorted(normalized):
+                connection.execute(
+                    sa.insert(references).values(
+                        id=str(uuid.uuid4()),
+                        tenant_id=row["tenant_id"],
+                        datasource_id=row["id"],
+                        relation=relation,
+                        reference=reference,
+                    )
+                )
@@ -9,6 +9,7 @@ from govoplan_core.core.datasources import (
    DatasourceAccessError,
    DatasourceDescriptor,
    DatasourceError,
+    DatasourceGovernance,
    DatasourceMaterialization,
    DatasourceNotFoundError,
    DatasourceOrigin,
@@ -23,6 +24,8 @@ from govoplan_datasources.backend.runtime import get_registry
 from govoplan_datasources.backend.schemas import (
    DatasourceFieldResponse,
    DatasourceFreezeRequest,
+    DatasourceGovernancePayload,
+    DatasourceGovernanceUpdateRequest,
    DatasourceListResponse,
    DatasourceMaterializationListResponse,
    DatasourceMaterializationResponse,
@@ -128,6 +131,7 @@ def api_register_origin(
            source_name=payload.source_name,
            mode=payload.mode,
            description=payload.description,
+            governance=_governance(payload.governance),
        )
    except DatasourceError as exc:
        raise _http_error(exc) from exc
@@ -204,6 +208,7 @@ def api_create_stage(
                    "source_format": payload.format,
                },
                metadata=payload.metadata,
+                governance=_governance(payload.governance),
            ),
        )
    except DatasourceError as exc:
@@ -270,6 +275,13 @@ def api_promote_stage(
 def api_list_datasources(
    query: str = Query(default="", max_length=200),
    limit: int = Query(default=100, ge=1, le=100),
+    authority_mode: str | None = Query(default=None, max_length=40),
+    classification: str | None = Query(default=None, max_length=80),
+    publication_state: str | None = Query(default=None, max_length=50),
+    owner_ref: str | None = Query(default=None, max_length=500),
+    responsible_organization_ref: str | None = Query(default=None, max_length=500),
+    affected_ref: str | None = Query(default=None, max_length=1000),
+    dependency_ref: str | None = Query(default=None, max_length=1000),
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
 ) -> DatasourceListResponse:
@@ -280,6 +292,13 @@ def api_list_datasources(
            principal,
            query=query,
            limit=limit,
+            authority_mode=authority_mode,
+            classification=classification,
+            publication_state=publication_state,
+            owner_ref=owner_ref,
+            responsible_organization_ref=responsible_organization_ref,
+            affected_ref=affected_ref,
+            dependency_ref=dependency_ref,
        )
    except DatasourceError as exc:
        raise _http_error(exc) from exc
@@ -311,6 +330,39 @@ def api_get_datasource(
    return _datasource_response(item)


+@router.patch(
+    "/{datasource_id}/governance",
+    response_model=DatasourceResponse,
+)
+def api_update_datasource_governance(
+    datasource_id: str,
+    payload: DatasourceGovernanceUpdateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> DatasourceResponse:
+    _require_any_scope(principal, SOURCE_WRITE_SCOPE, ADMIN_SCOPE)
+    try:
+        item = _provider().update_datasource_governance(
+            session,
+            principal,
+            datasource_ref=f"datasource:{datasource_id}",
+            governance=_governance(payload.governance)
+            or DatasourceGovernance(),
+        )
+    except DatasourceError as exc:
+        raise _http_error(exc) from exc
+    _audit(
+        session,
+        principal,
+        action="datasources.governance.updated",
+        object_type="datasource",
+        object_id=item.ref,
+        details={"governance": item.governance.to_dict()},
+    )
+    session.commit()
+    return _datasource_response(item)
+
+
@router.get(
    "/{datasource_id}/preview",
    response_model=DatasourcePreviewResponse,
@@ -509,6 +561,7 @@ def _datasource_response(item: DatasourceDescriptor) -> DatasourceResponse:
        capabilities=list(item.capabilities),
        provenance=dict(item.provenance),
        metadata=dict(item.metadata),
+        governance=item.governance.to_dict(),
    )


@@ -539,6 +592,7 @@ def _materialization_response(
        created_at=item.created_at.isoformat() if item.created_at else None,
        provenance=dict(item.provenance),
        metadata=dict(item.metadata),
+        governance=item.governance.to_dict(),
    )


@@ -569,6 +623,7 @@ def _stage_response(item: DatasourceStage) -> DatasourceStageResponse:
        promoted_materialization_ref=item.promoted_materialization_ref,
        provenance=dict(item.provenance),
        metadata=dict(item.metadata),
+        governance=item.governance.to_dict(),
    )


@@ -600,6 +655,14 @@ def _origin_response(item: DatasourceOrigin) -> DatasourceOriginResponse:
    )


+def _governance(
+    payload: DatasourceGovernancePayload | None,
+) -> DatasourceGovernance | None:
+    if payload is None:
+        return None
+    return DatasourceGovernance.from_mapping(payload.model_dump())
+
+
 def _audit(
    session: Session,
    principal: ApiPrincipal,
@@ -17,6 +17,52 @@ DatasourceKindValue = Literal[
    "custom",
 ]
 DatasourceShapeValue = Literal["tabular", "document", "binary", "directory", "stream"]
+SourceAuthorityModeValue = Literal[
+    "native_authoritative",
+    "external_authoritative",
+    "external_mirror",
+    "governed_sync",
+    "governance_overlay",
+    "linked_reference",
+]
+
+
+class DatasourceGovernancePayload(BaseModel):
+    owner_ref: str | None = Field(default=None, max_length=500)
+    steward_ref: str | None = Field(default=None, max_length=500)
+    responsible_organization_ref: str | None = Field(default=None, max_length=500)
+    responsible_function_ref: str | None = Field(default=None, max_length=500)
+    authoritative_source_ref: str | None = Field(default=None, max_length=1_000)
+    authority_mode: SourceAuthorityModeValue = "linked_reference"
+    legal_basis_refs: list[str] = Field(default_factory=list, max_length=100)
+    purposes: list[str] = Field(default_factory=list, max_length=100)
+    semantic_definition: str | None = Field(default=None, max_length=10_000)
+    schema_owner_ref: str | None = Field(default=None, max_length=500)
+    official_keys: list[str] = Field(default_factory=list, max_length=100)
+    classification: str = Field(default="internal", min_length=1, max_length=80)
+    privacy_profile_ref: str | None = Field(default=None, max_length=500)
+    retention_policy_ref: str | None = Field(default=None, max_length=500)
+    hold_refs: list[str] = Field(default_factory=list, max_length=100)
+    publication_state: str = Field(default="draft", min_length=1, max_length=50)
+    transfer_agreement_ref: str | None = Field(default=None, max_length=500)
+    freshness_policy: dict[str, Any] = Field(default_factory=dict)
+    quality_policy: dict[str, Any] = Field(default_factory=dict)
+    known_limits: list[str] = Field(default_factory=list, max_length=100)
+    correction_procedure_ref: str | None = Field(default=None, max_length=500)
+    affected_refs: list[str] = Field(default_factory=list, max_length=250)
+    dependency_refs: list[str] = Field(default_factory=list, max_length=250)
+
+    @model_validator(mode="after")
+    def external_source_is_explicit(self) -> "DatasourceGovernancePayload":
+        if (
+            self.authority_mode
+            in {"external_authoritative", "external_mirror", "governed_sync"}
+            and not str(self.authoritative_source_ref or "").strip()
+        ):
+            raise ValueError(
+                "External and synchronized datasources require an authoritative source reference."
+            )
+        return self


 class DatasourceFieldResponse(BaseModel):
@@ -46,6 +92,7 @@ class DatasourceResponse(BaseModel):
    capabilities: list[str]
    provenance: dict[str, Any]
    metadata: dict[str, Any]
+    governance: DatasourceGovernancePayload


 class DatasourceListResponse(BaseModel):
@@ -67,6 +114,7 @@ class DatasourceMaterializationResponse(BaseModel):
    created_at: str | None
    provenance: dict[str, Any]
    metadata: dict[str, Any]
+    governance: DatasourceGovernancePayload


 class DatasourceMaterializationListResponse(BaseModel):
@@ -92,6 +140,7 @@ class DatasourceStageResponse(BaseModel):
    promoted_materialization_ref: str | None
    provenance: dict[str, Any]
    metadata: dict[str, Any]
+    governance: DatasourceGovernancePayload


 class DatasourceStageListResponse(BaseModel):
@@ -112,6 +161,7 @@ class DatasourceStageCreateRequest(BaseModel):
    target_datasource_ref: str | None = None
    provenance: dict[str, Any] = Field(default_factory=dict)
    metadata: dict[str, Any] = Field(default_factory=dict)
+    governance: DatasourceGovernancePayload | None = None

    @model_validator(mode="after")
    def matching_payload(self) -> "DatasourceStageCreateRequest":
@@ -162,6 +212,11 @@ class DatasourceOriginRegisterRequest(BaseModel):
    source_name: str = Field(min_length=1, max_length=120)
    mode: Literal["live", "cached"]
    description: str | None = Field(default=None, max_length=4_000)
+    governance: DatasourceGovernancePayload | None = None
+
+
+class DatasourceGovernanceUpdateRequest(BaseModel):
+    governance: DatasourceGovernancePayload


 class DatasourcePreviewResponse(BaseModel):
@@ -7,7 +7,7 @@ from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, replace
 from typing import Any, cast

-from sqlalchemy import func, or_, select
+from sqlalchemy import exists, func, or_, select
 from sqlalchemy.orm import Session

 from govoplan_core.auth import ApiPrincipal, has_scope
@@ -16,6 +16,7 @@ from govoplan_core.core.datasources import (
    DatasourceDescriptor,
    DatasourceError,
    DatasourceField,
+    DatasourceGovernance,
    DatasourceMaterialization,
    DatasourceMode,
    DatasourceNotFoundError,
@@ -33,6 +34,7 @@ from govoplan_core.core.datasources import (
 )
 from govoplan_core.db.base import utcnow
 from govoplan_datasources.backend.db.models import (
+    DatasourceGovernanceReferenceRecord,
    DatasourceMaterializationRecord,
    DatasourcePayloadRecord,
    DatasourcePublicationRecord,
@@ -151,6 +153,13 @@ class SqlDatasourceProvider:
        *,
        query: str = "",
        limit: int = 100,
+        authority_mode: str | None = None,
+        classification: str | None = None,
+        publication_state: str | None = None,
+        owner_ref: str | None = None,
+        responsible_organization_ref: str | None = None,
+        affected_ref: str | None = None,
+        dependency_ref: str | None = None,
    ) -> Sequence[DatasourceDescriptor]:
        db, api_principal = _context(session, principal, CATALOGUE_READ_SCOPE)
        statement = (
@@ -172,6 +181,35 @@ class SqlDatasourceProvider:
                    DatasourceRecord.description.ilike(pattern, escape="\\"),
                )
            )
+        for column, value in (
+            (DatasourceRecord.authority_mode, authority_mode),
+            (DatasourceRecord.classification, classification),
+            (DatasourceRecord.publication_state, publication_state),
+            (DatasourceRecord.owner_ref, owner_ref),
+            (
+                DatasourceRecord.responsible_organization_ref,
+                responsible_organization_ref,
+            ),
+        ):
+            cleaned = str(value or "").strip()
+            if cleaned:
+                statement = statement.where(column == cleaned)
+        for relation, value in (
+            ("affected", affected_ref),
+            ("depends_on", dependency_ref),
+        ):
+            cleaned = str(value or "").strip()
+            if cleaned:
+                statement = statement.where(
+                    exists().where(
+                        DatasourceGovernanceReferenceRecord.datasource_id
+                        == DatasourceRecord.id,
+                        DatasourceGovernanceReferenceRecord.tenant_id
+                        == api_principal.tenant_id,
+                        DatasourceGovernanceReferenceRecord.relation == relation,
+                        DatasourceGovernanceReferenceRecord.reference == cleaned,
+                    )
+                )
        return tuple(_datasource_dto(item) for item in db.scalars(statement))

    def get_datasource(
@@ -378,6 +416,14 @@ class SqlDatasourceProvider:
                raise DatasourceValidationError(
                    "A stage can only update a datasource with the same mode and shape."
                )
+        governance = (
+            stage.governance
+            or (_datasource_governance(target) if target is not None else None)
+            or _default_governance(
+                mode=stage.mode,
+                provider_ref=stage.provider_ref,
+            )
+        )
        rows = normalize_rows(stage.rows)
        schema = infer_schema(rows)
        fingerprint = fingerprint_rows(rows, schema)
@@ -401,6 +447,7 @@ class SqlDatasourceProvider:
            validation_={"valid": True, "errors": [], "warnings": []},
            provenance_=dict(stage.provenance),
            metadata_=dict(stage.metadata),
+            governance_=governance.to_dict(),
            created_by=_actor_id(api_principal),
        )
        db.add(item)
@@ -461,6 +508,10 @@ class SqlDatasourceProvider:
                created_by=_actor_id(api_principal),
                updated_by=_actor_id(api_principal),
            )
+            _apply_datasource_governance(
+                datasource,
+                DatasourceGovernance.from_mapping(stage.governance_),
+            )
            db.add(datasource)
            db.flush()
            stage.target_datasource_id = datasource.id
@@ -468,6 +519,11 @@ class SqlDatasourceProvider:
            raise DatasourceValidationError(
                "A stage can only update a datasource with the same mode and shape."
            )
+        else:
+            _apply_datasource_governance(
+                datasource,
+                DatasourceGovernance.from_mapping(stage.governance_),
+            )

        materialization = _append_materialization(
            db,
@@ -502,6 +558,7 @@ class SqlDatasourceProvider:
        source_name: str,
        mode: DatasourceMode,
        description: str | None = None,
+        governance: DatasourceGovernance | None = None,
    ) -> DatasourceDescriptor:
        db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE)
        if mode not in {"live", "cached"}:
@@ -549,6 +606,11 @@ class SqlDatasourceProvider:
            created_by=_actor_id(api_principal),
            updated_by=_actor_id(api_principal),
        )
+        _apply_datasource_governance(
+            item,
+            governance
+            or _default_governance(mode=mode, provider_ref=origin.ref),
+        )
        db.add(item)
        db.flush()
        if mode == "cached":
@@ -567,6 +629,26 @@ class SqlDatasourceProvider:
            )
        return _datasource_dto(item)

+    def update_datasource_governance(
+        self,
+        session: object,
+        principal: object,
+        *,
+        datasource_ref: str,
+        governance: DatasourceGovernance,
+    ) -> DatasourceDescriptor:
+        db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE)
+        item = _required_datasource(
+            db,
+            tenant_id=api_principal.tenant_id,
+            datasource_ref=datasource_ref,
+            for_update=True,
+        )
+        _apply_datasource_governance(item, governance)
+        item.updated_by = _actor_id(api_principal)
+        db.flush()
+        return _datasource_dto(item)
+
    def refresh_datasource(
        self,
        session: object,
@@ -903,6 +985,7 @@ def _append_materialization(
        source_timestamp=source_timestamp,
        provenance_=dict(provenance or {}),
        metadata_=dict(metadata or {}),
+        governance_snapshot_=_datasource_governance(datasource).to_dict(),
        created_by=actor_id,
    )
    session.add(materialization)
@@ -1034,17 +1117,19 @@ def _datasource_record(
    *,
    tenant_id: str,
    datasource_ref: str,
+    for_update: bool = False,
 ) -> DatasourceRecord | None:
    datasource_id = _strip_ref(datasource_ref, "datasource:")
    if datasource_id is None:
        return None
-    return session.scalar(
-        select(DatasourceRecord).where(
+    statement = select(DatasourceRecord).where(
            DatasourceRecord.id == datasource_id,
            DatasourceRecord.tenant_id == tenant_id,
            DatasourceRecord.deleted_at.is_(None),
        )
-    )
+    if for_update:
+        statement = statement.with_for_update()
+    return session.scalar(statement)


 def _required_datasource(
@@ -1052,11 +1137,13 @@ def _required_datasource(
    *,
    tenant_id: str,
    datasource_ref: str,
+    for_update: bool = False,
 ) -> DatasourceRecord:
    item = _datasource_record(
        session,
        tenant_id=tenant_id,
        datasource_ref=datasource_ref,
+        for_update=for_update,
    )
    if item is None:
        raise DatasourceNotFoundError("Datasource not found.")
@@ -1120,6 +1207,96 @@ def _ensure_source_name_available(
        )


+def _default_governance(
+    *,
+    mode: str,
+    provider_ref: str | None,
+) -> DatasourceGovernance:
+    if mode == "live":
+        authority_mode = "external_authoritative"
+    elif mode == "cached" and provider_ref:
+        authority_mode = "external_mirror"
+    else:
+        authority_mode = "native_authoritative"
+    return DatasourceGovernance(
+        authoritative_source_ref=provider_ref,
+        authority_mode=cast(Any, authority_mode),
+        purposes=("governed_data_processing",),
+        publication_state="internal",
+    )
+
+
+def _datasource_governance(item: DatasourceRecord) -> DatasourceGovernance:
+    return DatasourceGovernance.from_mapping(
+        {
+            "owner_ref": item.owner_ref,
+            "steward_ref": item.steward_ref,
+            "responsible_organization_ref": item.responsible_organization_ref,
+            "responsible_function_ref": item.responsible_function_ref,
+            "authoritative_source_ref": item.authoritative_source_ref,
+            "authority_mode": item.authority_mode,
+            "legal_basis_refs": item.legal_basis_refs,
+            "purposes": item.purposes,
+            "semantic_definition": item.semantic_definition,
+            "schema_owner_ref": item.schema_owner_ref,
+            "official_keys": item.official_keys,
+            "classification": item.classification,
+            "privacy_profile_ref": item.privacy_profile_ref,
+            "retention_policy_ref": item.retention_policy_ref,
+            "hold_refs": item.hold_refs,
+            "publication_state": item.publication_state,
+            "transfer_agreement_ref": item.transfer_agreement_ref,
+            "freshness_policy": item.freshness_policy,
+            "quality_policy": item.quality_policy,
+            "known_limits": item.known_limits,
+            "correction_procedure_ref": item.correction_procedure_ref,
+            "affected_refs": item.affected_refs,
+            "dependency_refs": item.dependency_refs,
+        }
+    )
+
+
+def _apply_datasource_governance(
+    item: DatasourceRecord,
+    governance: DatasourceGovernance,
+) -> None:
+    item.owner_ref = governance.owner_ref
+    item.steward_ref = governance.steward_ref
+    item.responsible_organization_ref = governance.responsible_organization_ref
+    item.responsible_function_ref = governance.responsible_function_ref
+    item.authoritative_source_ref = governance.authoritative_source_ref
+    item.authority_mode = governance.authority_mode
+    item.legal_basis_refs = list(governance.legal_basis_refs)
+    item.purposes = list(governance.purposes)
+    item.semantic_definition = governance.semantic_definition
+    item.schema_owner_ref = governance.schema_owner_ref
+    item.official_keys = list(governance.official_keys)
+    item.classification = governance.classification
+    item.privacy_profile_ref = governance.privacy_profile_ref
+    item.retention_policy_ref = governance.retention_policy_ref
+    item.hold_refs = list(governance.hold_refs)
+    item.publication_state = governance.publication_state
+    item.transfer_agreement_ref = governance.transfer_agreement_ref
+    item.freshness_policy = dict(governance.freshness_policy)
+    item.quality_policy = dict(governance.quality_policy)
+    item.known_limits = list(governance.known_limits)
+    item.correction_procedure_ref = governance.correction_procedure_ref
+    item.affected_refs = list(governance.affected_refs)
+    item.dependency_refs = list(governance.dependency_refs)
+    item.governance_references = [
+        DatasourceGovernanceReferenceRecord(
+            tenant_id=item.tenant_id,
+            relation=relation,
+            reference=reference,
+        )
+        for relation, references in (
+            ("affected", governance.affected_refs),
+            ("depends_on", governance.dependency_refs),
+        )
+        for reference in references
+    ]
+
+
 def _datasource_dto(item: DatasourceRecord) -> DatasourceDescriptor:
    capabilities = ["read", "preview", "freeze"]
    if item.mode == "cached" and item.provider_ref:
@@ -1149,6 +1326,7 @@ def _datasource_dto(item: DatasourceRecord) -> DatasourceDescriptor:
        capabilities=tuple(capabilities),
        provenance=dict(item.provenance_),
        metadata=dict(item.metadata_),
+        governance=_datasource_governance(item),
    )


@@ -1170,6 +1348,7 @@ def _materialization_dto(
        created_at=item.created_at,
        provenance=dict(item.provenance_),
        metadata=dict(item.metadata_),
+        governance=DatasourceGovernance.from_mapping(item.governance_snapshot_),
    )


@@ -1201,6 +1380,7 @@ def _stage_dto(item: DatasourceStageRecord) -> DatasourceStage:
        ),
        provenance=dict(item.provenance_),
        metadata=dict(item.metadata_),
+        governance=DatasourceGovernance.from_mapping(item.governance_),
    )


@@ -1426,6 +1606,9 @@ def _publication_target(
            raise DatasourceValidationError(
                "Produced rows require a static or cached tabular datasource."
            )
+        if request.governance is not None:
+            _apply_datasource_governance(datasource, request.governance)
+            datasource.updated_by = actor_id
        return datasource
    name = str(request.name or "").strip()
    source_name = str(request.source_name or "").strip()
@@ -1468,6 +1651,11 @@ def _publication_target(
        created_by=actor_id,
        updated_by=actor_id,
    )
+    _apply_datasource_governance(
+        datasource,
+        request.governance
+        or _default_governance(mode="static", provider_ref=None),
+    )
    session.add(datasource)
    session.flush()
    return datasource