diff --git a/README.md b/README.md index 3f3b09a..4eee001 100644 --- a/README.md +++ b/README.md @@ -18,8 +18,14 @@ Datasource contracts rather than connector implementations. The first executable slice supports tabular static uploads, connector-backed live and cached sources, staging and promotion, refresh, immutable snapshots, -explicit frozen states, previews, and retirement. The contracts already model -database, HTTP/REST, directory, file, feed, document, binary, directory, and -stream sources so providers can be added without changing consumers. +explicit frozen states, previews, retirement, and atomic producer publication. +Producer modules can append a bounded tabular result or create a new static +datasource through an idempotent capability. The publication ledger retains the +producer run, output materialization, provenance, and replay identity. + +The contracts already model database, HTTP/REST, directory, file, feed, +document, binary, directory, and stream sources so providers can be added +without changing consumers. Larger durable artifact-backed publications remain +a later storage-provider slice. See [docs/CONCEPT.md](docs/CONCEPT.md) for ownership and lifecycle details. diff --git a/src/govoplan_datasources/backend/db/__init__.py b/src/govoplan_datasources/backend/db/__init__.py index 6d68dab..bd5e40e 100644 --- a/src/govoplan_datasources/backend/db/__init__.py +++ b/src/govoplan_datasources/backend/db/__init__.py @@ -2,12 +2,14 @@ from govoplan_datasources.backend.db.models import ( DatasourceMaterializationRecord, + DatasourcePublicationRecord, DatasourceRecord, DatasourceStageRecord, ) __all__ = [ "DatasourceMaterializationRecord", + "DatasourcePublicationRecord", "DatasourceRecord", "DatasourceStageRecord", ] diff --git a/src/govoplan_datasources/backend/db/models.py b/src/govoplan_datasources/backend/db/models.py index bf74d79..0103045 100644 --- a/src/govoplan_datasources/backend/db/models.py +++ b/src/govoplan_datasources/backend/db/models.py @@ -209,8 +209,66 @@ class DatasourceStageRecord(Base, TimestampMixin): created_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True) +class DatasourcePublicationRecord(Base, TimestampMixin): + __tablename__ = "datasource_publications" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "producer_module", + "idempotency_key", + name="uq_datasource_publication_idempotency", + ), + Index( + "ix_datasource_publications_producer_run", + "tenant_id", + "producer_module", + "producer_run_ref", + ), + Index( + "ix_datasource_publications_datasource", + "tenant_id", + "datasource_id", + ), + ) + + 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) + producer_module: Mapped[str] = mapped_column(String(100), nullable=False) + producer_run_ref: Mapped[str] = mapped_column(String(500), nullable=False) + idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False) + request_hash: Mapped[str] = mapped_column(String(64), nullable=False) + datasource_id: Mapped[str] = mapped_column( + ForeignKey("datasource_catalogue.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + materialization_id: Mapped[str] = mapped_column( + ForeignKey("datasource_materializations.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + status: Mapped[str] = mapped_column( + String(30), + default="published", + nullable=False, + index=True, + ) + details_: Mapped[dict[str, Any]] = mapped_column( + "details", + JSON, + default=dict, + nullable=False, + ) + created_by: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + index=True, + ) + + __all__ = [ "DatasourceMaterializationRecord", + "DatasourcePublicationRecord", "DatasourceRecord", "DatasourceStageRecord", "new_uuid", diff --git a/src/govoplan_datasources/backend/manifest.py b/src/govoplan_datasources/backend/manifest.py index 76c75b2..ca86734 100644 --- a/src/govoplan_datasources/backend/manifest.py +++ b/src/govoplan_datasources/backend/manifest.py @@ -10,6 +10,7 @@ from govoplan_core.core.datasources import ( CAPABILITY_DATASOURCE_CATALOGUE, CAPABILITY_DATASOURCE_LIFECYCLE, CAPABILITY_DATASOURCE_ORIGINS, + CAPABILITY_DATASOURCE_PUBLICATION, ) from govoplan_core.core.module_guards import ( drop_table_retirement_provider, @@ -181,6 +182,10 @@ manifest = ModuleManifest( name="datasources.staging", version=DATASOURCE_INTERFACE_VERSION, ), + ModuleInterfaceProvider( + name="datasources.publication", + version=DATASOURCE_INTERFACE_VERSION, + ), ), requires_interfaces=( ModuleInterfaceRequirement( @@ -218,6 +223,7 @@ manifest = ModuleManifest( capability_factories={ CAPABILITY_DATASOURCE_CATALOGUE: _provider, CAPABILITY_DATASOURCE_LIFECYCLE: _provider, + CAPABILITY_DATASOURCE_PUBLICATION: _provider, }, tenant_summary_providers=(_tenant_summary,), migration_spec=MigrationSpec( @@ -226,6 +232,7 @@ manifest = ModuleManifest( script_location=str(Path(__file__).with_name("migrations") / "versions"), retirement_supported=True, retirement_provider=drop_table_retirement_provider( + datasource_models.DatasourcePublicationRecord, datasource_models.DatasourceStageRecord, datasource_models.DatasourceMaterializationRecord, datasource_models.DatasourceRecord, @@ -241,6 +248,7 @@ manifest = ModuleManifest( datasource_models.DatasourceRecord, datasource_models.DatasourceMaterializationRecord, datasource_models.DatasourceStageRecord, + datasource_models.DatasourcePublicationRecord, label="Datasources", ), ), diff --git a/src/govoplan_datasources/backend/migrations/versions/c4e9f1a7d2b6_v0114_datasource_publications.py b/src/govoplan_datasources/backend/migrations/versions/c4e9f1a7d2b6_v0114_datasource_publications.py new file mode 100644 index 0000000..d602187 --- /dev/null +++ b/src/govoplan_datasources/backend/migrations/versions/c4e9f1a7d2b6_v0114_datasource_publications.py @@ -0,0 +1,132 @@ +"""v0.1.14 Datasource producer publications + +Revision ID: c4e9f1a7d2b6 +Revises: f3a8d2c7b1e0 +Create Date: 2026-07-28 00:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "c4e9f1a7d2b6" +down_revision = "f3a8d2c7b1e0" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "datasource_publications", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("producer_module", sa.String(length=100), nullable=False), + sa.Column("producer_run_ref", sa.String(length=500), nullable=False), + sa.Column("idempotency_key", sa.String(length=255), nullable=False), + sa.Column("request_hash", sa.String(length=64), nullable=False), + sa.Column("datasource_id", sa.String(length=36), nullable=False), + sa.Column("materialization_id", sa.String(length=36), nullable=False), + sa.Column("status", sa.String(length=30), nullable=False), + sa.Column("details", sa.JSON(), nullable=False), + sa.Column("created_by", sa.String(length=255), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["datasource_id"], + ["datasource_catalogue.id"], + name=op.f( + "fk_datasource_publications_datasource_id_datasource_catalogue" + ), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["materialization_id"], + ["datasource_materializations.id"], + name=op.f( + "fk_datasource_publications_materialization_id_datasource_materializations" + ), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_datasource_publications")), + sa.UniqueConstraint( + "tenant_id", + "producer_module", + "idempotency_key", + name="uq_datasource_publication_idempotency", + ), + ) + op.create_index( + op.f("ix_datasource_publications_created_by"), + "datasource_publications", + ["created_by"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_publications_datasource_id"), + "datasource_publications", + ["datasource_id"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_publications_materialization_id"), + "datasource_publications", + ["materialization_id"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_publications_status"), + "datasource_publications", + ["status"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_publications_tenant_id"), + "datasource_publications", + ["tenant_id"], + unique=False, + ) + op.create_index( + "ix_datasource_publications_datasource", + "datasource_publications", + ["tenant_id", "datasource_id"], + unique=False, + ) + op.create_index( + "ix_datasource_publications_producer_run", + "datasource_publications", + ["tenant_id", "producer_module", "producer_run_ref"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_index( + "ix_datasource_publications_producer_run", + table_name="datasource_publications", + ) + op.drop_index( + "ix_datasource_publications_datasource", + table_name="datasource_publications", + ) + op.drop_index( + op.f("ix_datasource_publications_tenant_id"), + table_name="datasource_publications", + ) + op.drop_index( + op.f("ix_datasource_publications_status"), + table_name="datasource_publications", + ) + op.drop_index( + op.f("ix_datasource_publications_materialization_id"), + table_name="datasource_publications", + ) + op.drop_index( + op.f("ix_datasource_publications_datasource_id"), + table_name="datasource_publications", + ) + op.drop_index( + op.f("ix_datasource_publications_created_by"), + table_name="datasource_publications", + ) + op.drop_table("datasource_publications") diff --git a/src/govoplan_datasources/backend/service.py b/src/govoplan_datasources/backend/service.py index 0752211..778b836 100644 --- a/src/govoplan_datasources/backend/service.py +++ b/src/govoplan_datasources/backend/service.py @@ -1,5 +1,8 @@ from __future__ import annotations +import hashlib +import json +import re from collections.abc import Mapping, Sequence from dataclasses import replace from typing import Any, cast @@ -18,6 +21,8 @@ from govoplan_core.core.datasources import ( DatasourceNotFoundError, DatasourceOrigin, DatasourceOriginReadRequest, + DatasourcePublicationRequest, + DatasourcePublicationResult, DatasourceReadRequest, DatasourceReadResult, DatasourceStage, @@ -29,6 +34,7 @@ from govoplan_core.core.datasources import ( from govoplan_core.db.base import utcnow from govoplan_datasources.backend.db.models import ( DatasourceMaterializationRecord, + DatasourcePublicationRecord, DatasourceRecord, DatasourceStageRecord, ) @@ -53,6 +59,178 @@ class SqlDatasourceProvider: def __init__(self, *, registry: object | None = None) -> None: self._registry = registry + def publish_rows( + self, + session: object, + principal: object, + *, + request: DatasourcePublicationRequest, + ) -> DatasourcePublicationResult: + db, api_principal = _publication_context(session, principal) + producer_module = request.producer_module.strip() + producer_run_ref = request.producer_run_ref.strip() + idempotency_key = request.idempotency_key.strip() + if not producer_module or len(producer_module) > 100: + raise DatasourceValidationError( + "A producer module of at most 100 characters is required." + ) + if not producer_run_ref or len(producer_run_ref) > 500: + raise DatasourceValidationError( + "A producer run reference of at most 500 characters is required." + ) + if not idempotency_key or len(idempotency_key) > 255: + raise DatasourceValidationError( + "An idempotency key of at most 255 characters is required." + ) + normalized = normalize_rows(request.rows) + schema = infer_schema(normalized) + fingerprint = fingerprint_rows(normalized, schema) + request_hash = _publication_request_hash( + request, + normalized=normalized, + fingerprint=fingerprint, + ) + existing = db.scalar( + select(DatasourcePublicationRecord).where( + DatasourcePublicationRecord.tenant_id + == api_principal.tenant_id, + DatasourcePublicationRecord.producer_module == producer_module, + DatasourcePublicationRecord.idempotency_key == idempotency_key, + ) + ) + if existing is not None: + if existing.request_hash != request_hash: + raise DatasourceValidationError( + "The publication idempotency key was already used with " + "different output." + ) + datasource = db.get(DatasourceRecord, existing.datasource_id) + materialization = db.get( + DatasourceMaterializationRecord, + existing.materialization_id, + ) + if ( + datasource is None + or materialization is None + or datasource.tenant_id != api_principal.tenant_id + or materialization.tenant_id != api_principal.tenant_id + ): + raise DatasourceUnavailableError( + "The prior publication result is no longer available." + ) + return DatasourcePublicationResult( + ref=_publication_ref(existing.id), + status=existing.status, + datasource=_datasource_dto(datasource), + materialization=_materialization_dto(materialization), + replayed=True, + ) + + datasource = None + if request.target_datasource_ref: + datasource = _required_datasource( + db, + tenant_id=api_principal.tenant_id, + datasource_ref=request.target_datasource_ref, + ) + if datasource.mode == "live" or datasource.shape != "tabular": + raise DatasourceValidationError( + "Produced rows require a static or cached tabular datasource." + ) + else: + name = str(request.name or "").strip() + source_name = str(request.source_name or "").strip() + if not name: + raise DatasourceValidationError( + "A datasource name is required for a new publication target." + ) + if not _valid_source_name(source_name): + raise DatasourceValidationError( + "Datasource keys must start with a letter or underscore and " + "contain only letters, numbers, and underscores." + ) + _ensure_source_name_available( + db, + tenant_id=api_principal.tenant_id, + source_name=source_name, + ) + datasource = DatasourceRecord( + tenant_id=api_principal.tenant_id, + source_name=source_name, + name=name, + description=_clean_optional(request.description), + kind="custom", + mode="static", + shape="tabular", + status="active", + provider=producer_module, + provider_ref=producer_run_ref, + schema_version=1, + schema_=[field_payload(field) for field in schema], + fingerprint=fingerprint, + row_count=len(normalized), + byte_count=encoded_size(normalized), + provenance_={ + **dict(request.provenance), + "producer_module": producer_module, + "producer_run_ref": producer_run_ref, + }, + metadata_=dict(request.metadata), + created_by=_actor_id(api_principal), + updated_by=_actor_id(api_principal), + ) + db.add(datasource) + db.flush() + + publication_provenance = { + **dict(request.provenance), + "producer_module": producer_module, + "producer_run_ref": producer_run_ref, + "idempotency_key": idempotency_key, + "published_at": utcnow().isoformat(), + } + materialization = _append_materialization( + db, + datasource=datasource, + rows=normalized, + schema=[field_payload(field) for field in schema], + fingerprint=fingerprint, + byte_count=encoded_size(normalized), + actor_id=_actor_id(api_principal), + frozen=request.freeze, + frozen_label=request.frozen_label, + source_timestamp=request.source_timestamp, + provenance=publication_provenance, + metadata=dict(request.metadata), + set_current=request.set_current, + ) + publication = DatasourcePublicationRecord( + tenant_id=api_principal.tenant_id, + producer_module=producer_module, + producer_run_ref=producer_run_ref, + idempotency_key=idempotency_key, + request_hash=request_hash, + datasource_id=datasource.id, + materialization_id=materialization.id, + status="published", + details_={ + "fingerprint": fingerprint, + "row_count": len(normalized), + "set_current": request.set_current, + "frozen": request.freeze, + }, + created_by=_actor_id(api_principal), + ) + db.add(publication) + db.flush() + return DatasourcePublicationResult( + ref=_publication_ref(publication.id), + status=publication.status, + datasource=_datasource_dto(datasource), + materialization=_materialization_dto(materialization), + replayed=False, + ) + def list_datasources( self, session: object, @@ -1038,6 +1216,19 @@ def _context( return session, principal +def _publication_context( + session: object, + principal: object, +) -> tuple[Session, ApiPrincipal]: + if not isinstance(session, Session): + raise TypeError("Datasource providers require a SQLAlchemy session.") + if not isinstance(principal, ApiPrincipal): + raise DatasourceAccessError("A tenant API principal is required.") + if not principal.tenant_id: + raise DatasourceAccessError("A tenant API principal is required.") + return session, principal + + def _validate_columns( schema: Sequence[Mapping[str, object]], columns: tuple[str, ...], @@ -1082,6 +1273,10 @@ def _stage_ref(item_id: str) -> str: return f"stage:{item_id}" +def _publication_ref(item_id: str) -> str: + return f"publication:{item_id}" + + def _actor_id(principal: ApiPrincipal) -> str | None: return principal.account_id or principal.membership_id or principal.identity_id @@ -1102,6 +1297,41 @@ def _escape_like(value: str) -> str: return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") +def _valid_source_name(value: str) -> bool: + return bool(re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,119}", value)) + + +def _publication_request_hash( + request: DatasourcePublicationRequest, + *, + normalized: Sequence[Mapping[str, object]], + fingerprint: str, +) -> str: + payload = { + "producer_module": request.producer_module.strip(), + "producer_run_ref": request.producer_run_ref.strip(), + "target_datasource_ref": request.target_datasource_ref, + "name": request.name, + "source_name": request.source_name, + "description": request.description, + "rows": [dict(row) for row in normalized], + "fingerprint": fingerprint, + "freeze": request.freeze, + "frozen_label": request.frozen_label, + "set_current": request.set_current, + "source_timestamp": request.source_timestamp, + "provenance": dict(request.provenance), + "metadata": dict(request.metadata), + } + encoded = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + default=str, + ) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + __all__ = [ "ADMIN_SCOPE", "CATALOGUE_READ_SCOPE", diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index abba19b..e43ff4c 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -14,12 +14,15 @@ from govoplan_core.core.datasources import ( DatasourceOrigin, DatasourceOriginReadRequest, DatasourceOriginReadResult, + DatasourcePublicationRequest, DatasourceReadRequest, DatasourceStageInput, + DatasourceValidationError, ) from govoplan_core.db.base import Base, utcnow from govoplan_datasources.backend.db.models import ( DatasourceMaterializationRecord, + DatasourcePublicationRecord, DatasourceRecord, DatasourceStageRecord, ) @@ -136,6 +139,7 @@ class DatasourceLifecycleTests(unittest.TestCase): DatasourceRecord.__table__, DatasourceMaterializationRecord.__table__, DatasourceStageRecord.__table__, + DatasourcePublicationRecord.__table__, ], ) self.Session = sessionmaker(bind=self.engine) @@ -151,6 +155,7 @@ class DatasourceLifecycleTests(unittest.TestCase): self.engine, tables=[ DatasourceStageRecord.__table__, + DatasourcePublicationRecord.__table__, DatasourceMaterializationRecord.__table__, DatasourceRecord.__table__, ], @@ -346,6 +351,85 @@ class DatasourceLifecycleTests(unittest.TestCase): tuple(item.ref for item in self.provider.list_stages(self.session, writer)), ) + def test_producer_publication_is_atomic_idempotent_and_addressable(self) -> None: + producer = principal(scopes=()) + request = DatasourcePublicationRequest( + producer_module="dataflow", + producer_run_ref="dataflow-run:run-1", + idempotency_key="monthly-output-2026-07", + name="Monthly comparison result", + source_name="monthly_comparison_result", + rows=( + {"case_id": "A-1", "result": "match"}, + {"case_id": "A-2", "result": "review"}, + ), + freeze=True, + frozen_label="July 2026", + provenance={"pipeline_revision": 3}, + ) + + first = self.provider.publish_rows( + self.session, + producer, + request=request, + ) + replay = self.provider.publish_rows( + self.session, + producer, + request=request, + ) + second = self.provider.publish_rows( + self.session, + producer, + request=DatasourcePublicationRequest( + producer_module="dataflow", + producer_run_ref="dataflow-run:run-2", + idempotency_key="monthly-output-2026-08", + target_datasource_ref=first.datasource.ref, + rows=({"case_id": "A-3", "result": "match"},), + ), + ) + self.session.commit() + + self.assertFalse(first.replayed) + self.assertTrue(replay.replayed) + self.assertEqual(first.ref, replay.ref) + self.assertEqual(first.materialization.ref, replay.materialization.ref) + self.assertIsNotNone(first.materialization.frozen_at) + self.assertEqual(1, first.materialization.revision) + self.assertEqual(2, second.materialization.revision) + self.assertEqual(first.datasource.ref, second.datasource.ref) + self.assertEqual( + 2, + self.session.query(DatasourcePublicationRecord).count(), + ) + + def test_publication_idempotency_key_rejects_different_output(self) -> None: + producer = principal(scopes=()) + base = DatasourcePublicationRequest( + producer_module="dataflow", + producer_run_ref="dataflow-run:run-1", + idempotency_key="stable-key", + name="Result", + source_name="result", + rows=({"id": 1},), + ) + self.provider.publish_rows(self.session, producer, request=base) + + with self.assertRaises(DatasourceValidationError): + self.provider.publish_rows( + self.session, + producer, + request=DatasourcePublicationRequest( + producer_module="dataflow", + producer_run_ref="dataflow-run:run-1", + idempotency_key="stable-key", + name="Result", + source_name="result", + rows=({"id": 2},), + ), + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_manifest.py b/tests/test_manifest.py index c140e37..1a6ed6b 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -5,6 +5,7 @@ import unittest from govoplan_core.core.datasources import ( CAPABILITY_DATASOURCE_CATALOGUE, CAPABILITY_DATASOURCE_LIFECYCLE, + CAPABILITY_DATASOURCE_PUBLICATION, ) from govoplan_datasources.backend.manifest import get_manifest @@ -16,6 +17,10 @@ class DatasourceManifestTests(unittest.TestCase): self.assertEqual(manifest.id, "datasources") self.assertIn(CAPABILITY_DATASOURCE_CATALOGUE, manifest.capability_factories) self.assertIn(CAPABILITY_DATASOURCE_LIFECYCLE, manifest.capability_factories) + self.assertIn( + CAPABILITY_DATASOURCE_PUBLICATION, + manifest.capability_factories, + ) self.assertIn( "datasources.materializations", {item.name for item in manifest.provides_interfaces}, diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 592c348..02b44c5 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -24,13 +24,14 @@ class DatasourceMigrationTests(unittest.TestCase): try: with engine.connect() as connection: self.assertIn( - "f3a8d2c7b1e0", + "c4e9f1a7d2b6", set(MigrationContext.configure(connection).get_current_heads()), ) self.assertEqual( { "datasource_catalogue", "datasource_materializations", + "datasource_publications", "datasource_stages", }, {