From 98d93e707f3666f708a30fa49fd894b2e9aa120f Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Sat, 1 Aug 2026 17:48:28 +0200 Subject: [PATCH] feat: implement governed datasource catalogue metadata --- AGENTS.md | 16 ++ docs/CONCEPT.md | 49 +++- src/govoplan_datasources/backend/db/models.py | 113 +++++++++ src/govoplan_datasources/backend/manifest.py | 66 +++++- ...2f6_v0114_governed_datasource_catalogue.py | 216 +++++++++++++++++ ...a0c3e7_v0114_governance_reference_index.py | 121 ++++++++++ src/govoplan_datasources/backend/router.py | 63 +++++ src/govoplan_datasources/backend/schemas.py | 55 +++++ src/govoplan_datasources/backend/service.py | 196 ++++++++++++++- tests/test_lifecycle.py | 106 +++++++++ tests/test_migrations.py | 99 +++++++- webui/src/api/datasources.ts | 62 ++++- .../features/datasources/DatasourcesPage.tsx | 223 ++++++++++++++++++ webui/src/styles/datasources.css | 9 + 14 files changed, 1382 insertions(+), 12 deletions(-) create mode 100644 AGENTS.md create mode 100644 src/govoplan_datasources/backend/migrations/versions/a7c1e4d9b2f6_v0114_governed_datasource_catalogue.py create mode 100644 src/govoplan_datasources/backend/migrations/versions/b8d2f5a0c3e7_v0114_governance_reference_index.py diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..17c0028 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,16 @@ +# GovOPlaN Datasources Codex Guide + +## Scope + +This repository owns the abstract datasource catalog, live/static/cached lifecycle, immutable snapshots, frozen states, staging references, and schema metadata. + +## Documentation Contract + +- Treat documentation as part of every behavior change. Update this module's manifest-driven `DocumentationTopic` contributions for affected user and administrator behavior. +- Keep feature content here; `govoplan-docs` projects it without importing Datasources internals. +- Maintain a static user/admin baseline and run `/mnt/DATA/git/govoplan/tools/checks/check-manifest-shapes.py` after behavior or manifest changes. + +## Boundaries + +- Connectors own external transport and credentials; Dataflow consumes governed datasource references. +- Preserve snapshot immutability, provenance, and scope-aware access. diff --git a/docs/CONCEPT.md b/docs/CONCEPT.md index 99421a8..e5833b8 100644 --- a/docs/CONCEPT.md +++ b/docs/CONCEPT.md @@ -2,7 +2,11 @@ ## Boundary -Datasources owns the governed identity and lifecycle of consumable data. +Datasources owns the governed identity and lifecycle of consumable data and +registers. It is not only a technical connection list: it is the catalogue in +which an institution explains what a dataset means, who is responsible for it, +why it may be used, how current and trustworthy it is, and which products or +decisions depend on it. Connectors owns external protocols, connection profiles, credentials, provider health, discovery, and source-side query pushdown. @@ -26,6 +30,45 @@ only retain opaque datasource and materialization references. Materializations are append-only. Changing source data creates a new revision; old revisions remain addressable for reproducibility. +The mode above describes how data is read. A separate authority declaration +states whether GovOPlaN is authoritative, an external source is authoritative, +the local state is a mirror, changes are governed in both directions, GovOPlaN +adds a governance overlay, or the entry is link-only. + +## Governance Catalogue + +A catalogue entry has typed, queryable governance fields for: + +- owner, steward, and responsible organization/function references; +- authoritative source, source-authority mode, and transfer/data-sharing + agreement references; +- legal or organizational basis, declared purposes, and permitted consumers; +- semantic definition, schema owner, official keys, and correction procedure; +- classification, privacy constraints, retention, legal hold, and publication + rules; +- freshness objective, quality policy, validation status, known limitations, + and incident state; +- affected services/processes and dependent Dataflows, reports, controls, + decisions, and published outputs. + +Authority mode, classification, publication state, owner, and responsible +organization are directly filterable catalogue columns. Lists and structured +quality/freshness rules retain typed API shapes. The governance editor is +available to datasource managers; readers see the effective explanation next +to the data preview. Each materialization captures the complete governance +state at publication time, so later ownership or policy edits do not rewrite +the explanation attached to prior evidence. + +Affected-object and dependency references are also normalized into an indexed, +tenant-scoped relation table while the original JSON lists remain readable for +compatibility. Catalogue clients can filter by exact `affected_ref` and +`dependency_ref` values without receiving row access to either the datasource +or the referenced object. + +Changing catalogue governance is audited. It does not rewrite source rows, +grant row access, or alter connector credentials. External and synchronized +authority modes require an explicit authoritative-source reference. + ## Staging Staging is the inspection boundary before data becomes generally consumable. @@ -49,6 +92,10 @@ Consumers request: Reads are bounded and tenant-scoped. Schema and expected fingerprints allow a Dataflow or Workflow definition to detect changed inputs before execution. +Consumers should be able to request the governance explanation and dependency +impact separately from row access. Seeing catalogue metadata must not imply +permission to read protected data. + ## Next Providers Connector providers should cover: diff --git a/src/govoplan_datasources/backend/db/models.py b/src/govoplan_datasources/backend/db/models.py index 76637d4..02791ab 100644 --- a/src/govoplan_datasources/backend/db/models.py +++ b/src/govoplan_datasources/backend/db/models.py @@ -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), diff --git a/src/govoplan_datasources/backend/manifest.py b/src/govoplan_datasources/backend/manifest.py index dd30e59..1b4eb7d 100644 --- a/src/govoplan_datasources/backend/manifest.py +++ b/src/govoplan_datasources/backend/manifest.py @@ -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"), diff --git a/src/govoplan_datasources/backend/migrations/versions/a7c1e4d9b2f6_v0114_governed_datasource_catalogue.py b/src/govoplan_datasources/backend/migrations/versions/a7c1e4d9b2f6_v0114_governed_datasource_catalogue.py new file mode 100644 index 0000000..fc163f1 --- /dev/null +++ b/src/govoplan_datasources/backend/migrations/versions/a7c1e4d9b2f6_v0114_governed_datasource_catalogue.py @@ -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"], + } + ) + ) diff --git a/src/govoplan_datasources/backend/migrations/versions/b8d2f5a0c3e7_v0114_governance_reference_index.py b/src/govoplan_datasources/backend/migrations/versions/b8d2f5a0c3e7_v0114_governance_reference_index.py new file mode 100644 index 0000000..12b6c25 --- /dev/null +++ b/src/govoplan_datasources/backend/migrations/versions/b8d2f5a0c3e7_v0114_governance_reference_index.py @@ -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, + ) + ) diff --git a/src/govoplan_datasources/backend/router.py b/src/govoplan_datasources/backend/router.py index 0d689d9..5454b76 100644 --- a/src/govoplan_datasources/backend/router.py +++ b/src/govoplan_datasources/backend/router.py @@ -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, diff --git a/src/govoplan_datasources/backend/schemas.py b/src/govoplan_datasources/backend/schemas.py index 4361fe3..f99744e 100644 --- a/src/govoplan_datasources/backend/schemas.py +++ b/src/govoplan_datasources/backend/schemas.py @@ -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): diff --git a/src/govoplan_datasources/backend/service.py b/src/govoplan_datasources/backend/service.py index fbac34e..d2178a1 100644 --- a/src/govoplan_datasources/backend/service.py +++ b/src/govoplan_datasources/backend/service.py @@ -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 diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index d3f7e97..46f2547 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -11,6 +11,7 @@ from govoplan_core.core.datasources import ( CAPABILITY_DATASOURCE_ORIGINS, DatasourceAccessError, DatasourceField, + DatasourceGovernance, DatasourceOrigin, DatasourceOriginReadRequest, DatasourceOriginReadResult, @@ -22,6 +23,7 @@ from govoplan_core.core.datasources import ( ) from govoplan_core.db.base import Base, utcnow from govoplan_datasources.backend.db.models import ( + DatasourceGovernanceReferenceRecord, DatasourceMaterializationRecord, DatasourcePayloadRecord, DatasourcePayloadRowRecord, @@ -146,6 +148,7 @@ class DatasourceLifecycleTests(unittest.TestCase): self.engine, tables=[ DatasourceRecord.__table__, + DatasourceGovernanceReferenceRecord.__table__, DatasourcePayloadRecord.__table__, DatasourcePayloadRowRecord.__table__, DatasourceMaterializationRecord.__table__, @@ -170,6 +173,7 @@ class DatasourceLifecycleTests(unittest.TestCase): DatasourceMaterializationRecord.__table__, DatasourcePayloadRowRecord.__table__, DatasourcePayloadRecord.__table__, + DatasourceGovernanceReferenceRecord.__table__, DatasourceRecord.__table__, ], ) @@ -261,6 +265,108 @@ class DatasourceLifecycleTests(unittest.TestCase): self.assertEqual(first_record.payload_id, frozen_record.payload_id) self.assertEqual([], first_record.rows) + def test_governance_is_queryable_and_snapshotted_per_materialization(self) -> None: + stage = self.provider.create_stage( + self.session, + principal(), + stage=DatasourceStageInput( + name="Governed register", + source_name="governed_register", + kind="upload", + mode="static", + shape="tabular", + rows=({"id": 1},), + governance=DatasourceGovernance( + owner_ref="function:data-owner", + steward_ref="account:steward", + responsible_organization_ref="organization:office-1", + authority_mode="native_authoritative", + legal_basis_refs=("policy:register-use",), + purposes=("case_processing",), + semantic_definition="Authoritative case register export.", + official_keys=("id",), + classification="restricted", + publication_state="internal", + quality_policy={"required_keys": ["id"]}, + affected_refs=("service:permit", "report:monthly"), + dependency_refs=("dataflow:monthly-case-check",), + ), + ), + ) + datasource, first = self.provider.promote_stage( + self.session, + principal(), + stage_ref=stage.ref, + ) + + self.assertEqual("function:data-owner", datasource.governance.owner_ref) + self.assertEqual("restricted", first.governance.classification) + self.assertEqual( + [datasource.ref], + [ + item.ref + for item in self.provider.list_datasources( + self.session, + principal(), + authority_mode="native_authoritative", + classification="restricted", + publication_state="internal", + owner_ref="function:data-owner", + responsible_organization_ref="organization:office-1", + ) + ], + ) + self.assertEqual( + [datasource.ref], + [ + item.ref + for item in self.provider.list_datasources( + self.session, + principal(), + affected_ref="service:permit", + dependency_ref="dataflow:monthly-case-check", + ) + ], + ) + self.assertEqual( + (), + self.provider.list_datasources( + self.session, + principal("tenant-2"), + affected_ref="service:permit", + ), + ) + + changed = self.provider.update_datasource_governance( + self.session, + principal(), + datasource_ref=datasource.ref, + governance=DatasourceGovernance( + owner_ref="function:new-owner", + authority_mode="native_authoritative", + purposes=("case_processing",), + classification="confidential", + publication_state="internal", + ), + ) + history = self.provider.list_materializations( + self.session, + principal(), + datasource_ref=datasource.ref, + ) + + self.assertEqual("function:new-owner", changed.governance.owner_ref) + self.assertEqual("function:data-owner", history[0].governance.owner_ref) + self.assertEqual("restricted", history[0].governance.classification) + self.assertEqual( + (), + self.provider.list_datasources( + self.session, + principal(), + dependency_ref="dataflow:monthly-case-check", + ), + ) + def test_live_reads_origin_and_cached_refresh_is_explicit(self) -> None: live = self.provider.register_origin( self.session, diff --git a/tests/test_migrations.py b/tests/test_migrations.py index d1c6467..703afc9 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -2,16 +2,26 @@ from __future__ import annotations import tempfile import unittest +from datetime import UTC, datetime from pathlib import Path +from alembic import command from alembic.runtime.migration import MigrationContext -from sqlalchemy import create_engine, inspect +from sqlalchemy import MetaData, Table, create_engine, inspect, select -from govoplan_core.db.migrations import migrate_database +from govoplan_core.db.migrations import alembic_config, migrate_database from govoplan_datasources.backend.manifest import get_manifest class DatasourceMigrationTests(unittest.TestCase): + @staticmethod + def _config(url: str): + return alembic_config( + database_url=url, + enabled_modules=("datasources",), + manifest_factories=(get_manifest,), + ) + def test_baseline_creates_datasource_tables_and_head(self) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-datasources-migration-") as directory: url = f"sqlite:///{Path(directory) / 'datasources.db'}" @@ -24,12 +34,29 @@ class DatasourceMigrationTests(unittest.TestCase): try: with engine.connect() as connection: self.assertIn( - "d5f0a2b8c3e7", + "b8d2f5a0c3e7", set(MigrationContext.configure(connection).get_current_heads()), ) + catalogue_columns = { + item["name"] + for item in inspect(connection).get_columns( + "datasource_catalogue" + ) + } + self.assertTrue( + { + "authority_mode", + "classification", + "publication_state", + "owner_ref", + "quality_policy", + "dependency_refs", + }.issubset(catalogue_columns) + ) self.assertEqual( { "datasource_catalogue", + "datasource_governance_references", "datasource_materializations", "datasource_payload_rows", "datasource_payloads", @@ -45,6 +72,72 @@ class DatasourceMigrationTests(unittest.TestCase): finally: engine.dispose() + def test_governance_reference_index_backfills_existing_catalogue_rows( + self, + ) -> None: + with tempfile.TemporaryDirectory( + prefix="govoplan-datasources-governance-migration-" + ) as directory: + url = f"sqlite:///{Path(directory) / 'datasources.db'}" + config = self._config(url) + command.upgrade(config, "a7c1e4d9b2f6") + engine = create_engine(url) + try: + metadata = MetaData() + catalogue = Table( + "datasource_catalogue", metadata, autoload_with=engine + ) + now = datetime.now(UTC) + with engine.begin() as connection: + connection.execute( + catalogue.insert().values( + id="datasource-1", + tenant_id="tenant-1", + source_name="governed", + name="Governed", + kind="upload", + mode="static", + shape="tabular", + status="active", + schema_version=1, + schema=[], + fingerprint="", + provenance={}, + metadata={}, + affected_refs=["service:permit", "service:permit"], + dependency_refs=["dataflow:monthly"], + created_at=now, + updated_at=now, + ) + ) + + command.upgrade(config, "b8d2f5a0c3e7") + + references = Table( + "datasource_governance_references", + MetaData(), + autoload_with=engine, + ) + with engine.connect() as connection: + rows = connection.execute( + select( + references.c.relation, + references.c.reference, + ).order_by( + references.c.relation, + references.c.reference, + ) + ).all() + self.assertEqual( + [ + ("affected", "service:permit"), + ("depends_on", "dataflow:monthly"), + ], + rows, + ) + finally: + engine.dispose() + if __name__ == "__main__": unittest.main() diff --git a/webui/src/api/datasources.ts b/webui/src/api/datasources.ts index 40fb90f..e586481 100644 --- a/webui/src/api/datasources.ts +++ b/webui/src/api/datasources.ts @@ -5,6 +5,39 @@ import { export type DatasourceMode = "live" | "cached" | "static"; export type DatasourceShape = "tabular" | "document" | "binary" | "directory" | "stream"; +export type SourceAuthorityMode = + | "native_authoritative" + | "external_authoritative" + | "external_mirror" + | "governed_sync" + | "governance_overlay" + | "linked_reference"; + +export type DatasourceGovernance = { + owner_ref?: string | null; + steward_ref?: string | null; + responsible_organization_ref?: string | null; + responsible_function_ref?: string | null; + authoritative_source_ref?: string | null; + authority_mode: SourceAuthorityMode; + legal_basis_refs: string[]; + purposes: string[]; + semantic_definition?: string | null; + schema_owner_ref?: string | null; + official_keys: string[]; + classification: string; + privacy_profile_ref?: string | null; + retention_policy_ref?: string | null; + hold_refs: string[]; + publication_state: string; + transfer_agreement_ref?: string | null; + freshness_policy: Record; + quality_policy: Record; + known_limits: string[]; + correction_procedure_ref?: string | null; + affected_refs: string[]; + dependency_refs: string[]; +}; export type DatasourceField = { name: string; @@ -33,6 +66,7 @@ export type Datasource = { capabilities: string[]; provenance: Record; metadata: Record; + governance: DatasourceGovernance; }; export type DatasourceMaterialization = { @@ -50,6 +84,7 @@ export type DatasourceMaterialization = { created_at?: string | null; provenance: Record; metadata: Record; + governance: DatasourceGovernance; }; export type DatasourceStage = { @@ -71,6 +106,7 @@ export type DatasourceStage = { promoted_materialization_ref?: string | null; provenance: Record; metadata: Record; + governance: DatasourceGovernance; }; export type DatasourceOrigin = { @@ -102,9 +138,18 @@ export type DatasourcePreview = { export async function listDatasources( settings: ApiSettings, - query = "" + query = "", + filters: Partial> & { + affected_ref?: string; + dependency_ref?: string; + } = {} ): Promise { - const suffix = query.trim() ? `?query=${encodeURIComponent(query.trim())}` : ""; + const params = new URLSearchParams(); + if (query.trim()) params.set("query", query.trim()); + for (const [key, value] of Object.entries(filters)) { + if (String(value ?? "").trim()) params.set(key, String(value).trim()); + } + const suffix = params.size ? `?${params.toString()}` : ""; const response = await apiFetch<{ datasources: Datasource[] }>( settings, `/api/v1/datasources${suffix}` @@ -158,6 +203,7 @@ export function createDatasourceStage( description?: string | null; mode: "static" | "cached"; target_datasource_ref?: string | null; + governance?: DatasourceGovernance | null; } & ( { format: "json"; rows: Record[] } | { format: "csv"; csv_text: string; delimiter: string } @@ -188,6 +234,7 @@ export function registerDatasourceOrigin( source_name: string; mode: "live" | "cached"; description?: string | null; + governance?: DatasourceGovernance | null; } ): Promise { return apiFetch(settings, "/api/v1/datasources/origins/register", { @@ -225,6 +272,17 @@ export function retireDatasource( }); } +export function updateDatasourceGovernance( + settings: ApiSettings, + datasourceRef: string, + governance: DatasourceGovernance +): Promise { + return apiFetch(settings, `/api/v1/datasources/${refId(datasourceRef)}/governance`, { + method: "PATCH", + body: JSON.stringify({ governance }) + }); +} + function refId(ref: string): string { const separator = ref.indexOf(":"); return encodeURIComponent(separator >= 0 ? ref.slice(separator + 1) : ref); diff --git a/webui/src/features/datasources/DatasourcesPage.tsx b/webui/src/features/datasources/DatasourcesPage.tsx index 83ccb98..3eb3bc7 100644 --- a/webui/src/features/datasources/DatasourcesPage.tsx +++ b/webui/src/features/datasources/DatasourcesPage.tsx @@ -11,9 +11,11 @@ import { Download, Eye, Layers3, + Pencil, Plus, RefreshCw, Snowflake, + ShieldCheck, Trash2, Upload } from "lucide-react"; @@ -44,7 +46,9 @@ import { refreshDatasource, registerDatasourceOrigin, retireDatasource, + updateDatasourceGovernance, type Datasource, + type DatasourceGovernance, type DatasourceMaterialization, type DatasourceOrigin, type DatasourcePreview, @@ -82,6 +86,7 @@ export default function DatasourcesPage({ const [freezeOpen, setFreezeOpen] = useState(false); const [freezeLabel, setFreezeLabel] = useState(""); const [retireOpen, setRetireOpen] = useState(false); + const [governanceOpen, setGovernanceOpen] = useState(false); const canManage = hasScope(auth, "datasources:source:write") || hasScope(auth, "datasources:source:admin"); @@ -386,6 +391,12 @@ export default function DatasourcesPage({ ) : null} {view === "catalogue" && selectedDatasource ? ( <> + } + onClick={() => setGovernanceOpen(true)} + disabled={!canManage || working} + /> @@ -485,6 +496,17 @@ export default function DatasourcesPage({ } }} /> + setGovernanceOpen(false)} + onSaved={async (updated) => { + setGovernanceOpen(false); + setSuccess(`Updated governance for ${updated.name}.`); + await reload(updated.ref); + }} + /> {datasource.description} ) : null} +
+
+ Governance + +
+
+ Authority{readableToken(datasource.governance.authority_mode)} + Classification{datasource.governance.classification} + Owner{datasource.governance.owner_ref || "Not assigned"} + Steward{datasource.governance.steward_ref || "Not assigned"} + Responsible organization{datasource.governance.responsible_organization_ref || "Not assigned"} + Purposes{datasource.governance.purposes.join(", ") || "Not declared"} +
+ {datasource.governance.semantic_definition ? ( +

{datasource.governance.semantic_definition}

+ ) : null} +
Preview @@ -684,6 +726,173 @@ function OriginDetail({ origin }: { origin: DatasourceOrigin }) { ); } +function GovernanceDialog({ + open, + settings, + datasource, + onClose, + onSaved +}: { + open: boolean; + settings: ApiSettings; + datasource: Datasource | null; + onClose: () => void; + onSaved: (datasource: Datasource) => void | Promise; +}) { + const [draft, setDraft] = useState(null); + const [freshness, setFreshness] = useState("{}"); + const [quality, setQuality] = useState("{}"); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(""); + + useEffect(() => { + if (!open || !datasource) return; + setDraft(structuredClone(datasource.governance)); + setFreshness(JSON.stringify(datasource.governance.freshness_policy, null, 2)); + setQuality(JSON.stringify(datasource.governance.quality_policy, null, 2)); + setError(""); + }, [datasource, open]); + + const save = async () => { + if (!datasource || !draft) return; + setBusy(true); + setError(""); + try { + const updated = await updateDatasourceGovernance(settings, datasource.ref, { + ...draft, + freshness_policy: parseObject(freshness, "Freshness policy"), + quality_policy: parseObject(quality, "Quality policy") + }); + await onSaved(updated); + } catch (saveError) { + setError(apiErrorMessage(saveError)); + } finally { + setBusy(false); + } + }; + + const setValue = ( + key: K, + value: DatasourceGovernance[K] + ) => setDraft((current) => current ? { ...current, [key]: value } : current); + + return ( + { if (!busy) onClose(); }} + footer={( + <> + + + + )} + > + {error ? {error} : null} + {draft ? ( +
+
+ + + + + setValue("authoritative_source_ref", event.target.value || null)} /> + + + setValue("classification", event.target.value)} /> + + + setValue("publication_state", event.target.value)} /> + + + setValue("owner_ref", event.target.value || null)} /> + + + setValue("steward_ref", event.target.value || null)} /> + + + setValue("responsible_organization_ref", event.target.value || null)} /> + + + setValue("responsible_function_ref", event.target.value || null)} /> + + + setValue("schema_owner_ref", event.target.value || null)} /> + + + setValue("privacy_profile_ref", event.target.value || null)} /> + + + setValue("retention_policy_ref", event.target.value || null)} /> + + + setValue("transfer_agreement_ref", event.target.value || null)} /> + + + setValue("correction_procedure_ref", event.target.value || null)} /> + +
+ +