diff --git a/docs/QUALITY_POLICY.md b/docs/QUALITY_POLICY.md index 25bbe3a..be55911 100644 --- a/docs/QUALITY_POLICY.md +++ b/docs/QUALITY_POLICY.md @@ -1,10 +1,10 @@ -# Datasource Stage Quality Policy +# Datasource Quality Policy Datasource quality policy is a deterministic JSON contract stored in -`governance.quality_policy`. A tabular stage inherits the current target -Datasource policy unless the stage has its own governed definition. Validation -runs before the stage is stored, but a failed stage remains available for -inspection and correction through a new stage. +`governance.quality_policy`. A tabular stage or producer publication inherits +the current target Datasource policy unless it supplies its own governed +definition. Stage validation runs before the stage is stored, but a failed +stage remains available for inspection and correction through a new stage. ## Contract @@ -78,6 +78,18 @@ contain only affected counts and at most 25 one-based row numbers; they never copy field values. Promotion copies this validation object into the immutable materialization provenance and records the policy hash in the audit event. +Producer publication uses the same gate before any catalogue target, +materialization, or publication record is persisted. A rejected output has no +partial catalogue effect. Successful output materializations retain the exact +validation result, policy version/hash, and schema classification in +`provenance.publication_validation`; the publication record retains the same +evidence for operational inspection. Dataflow, Workflow, and Reporting can +therefore consume an immutable output reference without re-running a possibly +changed quality policy. Publication also emits a transactional +`datasource.publication.published` audit/platform event; an enabled Audit +module stores it in the durable outbox, while reduced installations deliver it +through Core after commit. + Approval authority, approval expiry, and retention/deletion execution remain separate work under `govoplan-datasources#2`. Until those contracts are added, no JSON flag is treated as an approval and no stage is deleted automatically. diff --git a/src/govoplan_datasources/backend/manifest.py b/src/govoplan_datasources/backend/manifest.py index 64cb954..3a48f6b 100644 --- a/src/govoplan_datasources/backend/manifest.py +++ b/src/govoplan_datasources/backend/manifest.py @@ -406,15 +406,16 @@ manifest = ModuleManifest( ), DocumentationTopic( id="datasources.quality-gates", - title="Validate a Datasource stage", - summary="Apply deterministic quality rules and schema-change policy before staged data can become consumable.", + title="Validate staged and produced Datasource revisions", + summary="Apply deterministic quality rules and schema-change policy before staged or produced data can become consumable.", body=( "Tabular stages evaluate configured row count, required-field, field-shape, nullability, uniqueness, numeric-range, and bounded " "referential-set rules. Errors keep the stage inspectable but block promotion; warnings stay visible and permit an explicit " "promotion. Updates compare the detected schema with the current target and classify each change as compatible, warning, or " "breaking. Diagnostics expose counts and bounded row numbers, never field values. The policy version and hash, diagnostics, and " - "schema diff are copied into immutable materialization provenance when promotion succeeds. Approval and retention execution are " - "not inferred from arbitrary JSON flags and remain separate governed lifecycle work." + "schema diff are copied into immutable materialization provenance when promotion succeeds. Producer publication applies the same " + "gate before any catalogue effect, retains validation evidence on the immutable output revision, and emits a transactional terminal event. Approval and retention execution " + "are not inferred from arbitrary JSON flags and remain separate governed lifecycle work." ), layer="available", documentation_types=("admin", "user"), diff --git a/src/govoplan_datasources/backend/service.py b/src/govoplan_datasources/backend/service.py index 945da97..8269636 100644 --- a/src/govoplan_datasources/backend/service.py +++ b/src/govoplan_datasources/backend/service.py @@ -10,6 +10,7 @@ from typing import Any, cast from sqlalchemy import exists, func, or_, select from sqlalchemy.orm import Session +from govoplan_core.audit.logging import audit_event from govoplan_core.auth import ApiPrincipal, has_scope from govoplan_core.core.datasources import ( DatasourceAccessError, @@ -108,12 +109,37 @@ class SqlDatasourceProvider: return existing actor_id = _actor_id(api_principal) + target, governance, baseline_schema = _publication_validation_context( + db, + tenant_id=api_principal.tenant_id, + request=request, + ) + validation = validate_stage( + rows=prepared.rows, + schema=prepared.schema, + quality_policy=governance.quality_policy, + baseline_schema=baseline_schema, + ) + if validation["valid"] is not True: + errors = validation.get("errors", []) + error_codes = ", ".join( + str(item.get("code") or "validation.error") + for item in errors + if isinstance(item, Mapping) + ) + suffix = f" ({error_codes})" if error_codes else "" + raise DatasourceValidationError( + "Published output failed governed quality or schema validation" + f"{suffix}." + ) datasource = _publication_target( db, tenant_id=api_principal.tenant_id, actor_id=actor_id, request=request, prepared=prepared, + target=target, + governance=governance, ) materialization = _append_materialization( db, @@ -126,7 +152,10 @@ class SqlDatasourceProvider: frozen=request.freeze, frozen_label=request.frozen_label, source_timestamp=request.source_timestamp, - provenance=_publication_provenance(request, prepared), + provenance={ + **_publication_provenance(request, prepared), + "publication_validation": validation, + }, metadata=dict(request.metadata), set_current=request.set_current, ) @@ -139,6 +168,28 @@ class SqlDatasourceProvider: request=request, prepared=prepared, ) + audit_event( + db, + tenant_id=api_principal.tenant_id, + user_id=getattr(api_principal.user, "id", None) + or api_principal.account_id, + api_key_id=api_principal.api_key_id, + action="datasource.publication.published", + object_type="datasource_publication", + object_id=publication.id, + details={ + "producer_module": prepared.producer_module, + "producer_run_ref": prepared.producer_run_ref, + "datasource_ref": _datasource_ref(datasource.id), + "materialization_ref": _materialization_ref(materialization.id), + "fingerprint": prepared.fingerprint, + "row_count": len(prepared.rows), + "policy_hash": validation["policy_hash"], + "schema_classification": validation["schema_change"][ + "classification" + ], + }, + ) return DatasourcePublicationResult( ref=_publication_ref(publication.id), status=publication.status, @@ -1607,19 +1658,13 @@ def _publication_target( actor_id: str | None, request: DatasourcePublicationRequest, prepared: _PreparedPublication, + target: DatasourceRecord | None, + governance: DatasourceGovernance, ) -> DatasourceRecord: - if request.target_datasource_ref: - datasource = _required_datasource( - session, - tenant_id=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." - ) + if target is not None: + datasource = target if request.governance is not None: - _apply_datasource_governance(datasource, request.governance) + _apply_datasource_governance(datasource, governance) datasource.updated_by = actor_id return datasource name = str(request.name or "").strip() @@ -1665,14 +1710,47 @@ def _publication_target( ) _apply_datasource_governance( datasource, - request.governance - or _default_governance(mode="static", provider_ref=None), + governance, ) session.add(datasource) session.flush() return datasource +def _publication_validation_context( + session: Session, + *, + tenant_id: str, + request: DatasourcePublicationRequest, +) -> tuple[ + DatasourceRecord | None, + DatasourceGovernance, + tuple[DatasourceField, ...] | None, +]: + if not request.target_datasource_ref: + return ( + None, + request.governance + or _default_governance(mode="static", provider_ref=None), + None, + ) + target = _required_datasource( + session, + tenant_id=tenant_id, + datasource_ref=request.target_datasource_ref, + for_update=True, + ) + if target.mode == "live" or target.shape != "tabular": + raise DatasourceValidationError( + "Produced rows require a static or cached tabular datasource." + ) + return ( + target, + request.governance or _datasource_governance(target), + _fields(target.schema_) if target.current_materialization_id else None, + ) + + def _publication_provenance( request: DatasourcePublicationRequest, prepared: _PreparedPublication, @@ -1710,6 +1788,10 @@ def _create_publication_record( "row_count": len(prepared.rows), "set_current": request.set_current, "frozen": request.freeze, + "validation": materialization.provenance_.get( + "publication_validation", + {}, + ), }, created_by=actor_id, ) diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index 1e7099c..f506050 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -7,6 +7,7 @@ from sqlalchemy.orm import sessionmaker from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.core.datasources import ( CAPABILITY_DATASOURCE_ORIGINS, DatasourceAccessError, @@ -154,6 +155,7 @@ class DatasourceLifecycleTests(unittest.TestCase): DatasourceMaterializationRecord.__table__, DatasourceStageRecord.__table__, DatasourcePublicationRecord.__table__, + ChangeSequenceEntry.__table__, ], ) self.Session = sessionmaker(bind=self.engine) @@ -175,6 +177,7 @@ class DatasourceLifecycleTests(unittest.TestCase): DatasourcePayloadRecord.__table__, DatasourceGovernanceReferenceRecord.__table__, DatasourceRecord.__table__, + ChangeSequenceEntry.__table__, ], ) self.engine.dispose() @@ -651,6 +654,106 @@ class DatasourceLifecycleTests(unittest.TestCase): ), ) + def test_publication_enforces_quality_before_persisting_any_effect(self) -> None: + producer = principal(scopes=(SOURCE_WRITE_SCOPE,)) + blocked = DatasourcePublicationRequest( + producer_module="dataflow", + producer_run_ref="dataflow-run:blocked", + idempotency_key="blocked-output", + name="Blocked output", + source_name="blocked_output", + rows=({"id": 1},), + governance=DatasourceGovernance( + quality_policy={ + "version": "minimum-two-v1", + "rules": [ + { + "id": "minimum-two", + "type": "row_count", + "minimum": 2, + } + ], + }, + ), + ) + + with self.assertRaisesRegex( + DatasourceValidationError, + r"quality or schema validation \(quality\.row_count\)", + ): + self.provider.publish_rows(self.session, producer, request=blocked) + + self.assertEqual(0, self.session.query(DatasourceRecord).count()) + self.assertEqual( + 0, + self.session.query(DatasourceMaterializationRecord).count(), + ) + self.assertEqual( + 0, + self.session.query(DatasourcePublicationRecord).count(), + ) + + def test_publication_preserves_quality_and_schema_evidence(self) -> None: + producer = principal(scopes=(SOURCE_WRITE_SCOPE,)) + first = self.provider.publish_rows( + self.session, + producer, + request=DatasourcePublicationRequest( + producer_module="dataflow", + producer_run_ref="dataflow-run:first", + idempotency_key="governed-output-first", + name="Governed output", + source_name="governed_output", + rows=({"id": 1},), + governance=DatasourceGovernance( + quality_policy={ + "version": "governed-output-v1", + "rules": [ + { + "id": "unique-id", + "type": "unique", + "fields": ["id"], + }, + ], + "schema_policy": {"field_added_required": "warning"}, + }, + ), + ), + ) + second = self.provider.publish_rows( + self.session, + producer, + request=DatasourcePublicationRequest( + producer_module="dataflow", + producer_run_ref="dataflow-run:second", + idempotency_key="governed-output-second", + target_datasource_ref=first.datasource.ref, + rows=({"id": 2, "result": "match"},), + ), + ) + + validation = second.materialization.provenance["publication_validation"] + self.assertTrue(validation["valid"]) + self.assertEqual("governed-output-v1", validation["policy_version"]) + self.assertEqual( + "warning", + validation["schema_change"]["classification"], + ) + self.assertEqual( + "schema.field_added_required", + validation["warnings"][0]["code"], + ) + record = self.session.get( + DatasourcePublicationRecord, + second.ref.removeprefix("publication:"), + ) + self.assertIsNotNone(record) + assert record is not None + self.assertEqual( + validation["policy_hash"], + record.details_["validation"]["policy_hash"], + ) + def test_publication_requires_source_write_scope(self) -> None: with self.assertRaises(DatasourceAccessError): self.provider.publish_rows(