From 4e0238f70175f13c41f5749c598ef13009986e6f Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 12:48:37 +0200 Subject: [PATCH] Pin reporting datasets to published Dataflow runs --- docs/ADMIN_GUIDE.md | 4 ++ docs/USER_GUIDE.md | 5 ++ src/govoplan_reporting/backend/execution.py | 1 + src/govoplan_reporting/backend/manifest.py | 2 +- src/govoplan_reporting/backend/schemas.py | 3 + tests/test_reporting_service.py | 70 +++++++++++++++++++++ 6 files changed, 84 insertions(+), 1 deletion(-) diff --git a/docs/ADMIN_GUIDE.md b/docs/ADMIN_GUIDE.md index 13cc9d4..753eb26 100644 --- a/docs/ADMIN_GUIDE.md +++ b/docs/ADMIN_GUIDE.md @@ -27,6 +27,10 @@ published by a source-owning module. Do not expose another module's ORM or an unbounded SQL connection as a report source. Configure an explicit schema, freshness policy, source fingerprint expectations, purpose, privacy, retention, and a row-policy provider where source access alone is not enough. +For a Dataflow source, `source_run_ref` optionally pins one successful run that +published an immutable Datasource materialization. Reporting passes this pin +through rather than re-executing the pipeline. The current principal must still +be authorized for the Dataflow definition and the exact Datasource output. On PostgreSQL installations, Reporting compiles bounded semantic filters, grouping, measures, calculated measures, ordering, offsets, and limits into a diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index 0951e74..f23e479 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -28,6 +28,11 @@ the exact definition and output. Warnings explain freshness, inferred schema, or provider diagnostics. A failed quality gate records a failed execution and does not publish a result. +A report configured against an exact published Dataflow run reads that run's +immutable Datasource materialization. It does not rerun the flow with current +inputs. The evidence identifies the Dataflow run and materialization, and access +to both is checked again when the report runs. + The **Effective access** explanation states when dimensions, measures, source rows, or actions were removed by Policy. A result with no hidden elements says so explicitly; catalogue visibility never grants access to protected detail. diff --git a/src/govoplan_reporting/backend/execution.py b/src/govoplan_reporting/backend/execution.py index 8d8544a..a383a20 100644 --- a/src/govoplan_reporting/backend/execution.py +++ b/src/govoplan_reporting/backend/execution.py @@ -501,6 +501,7 @@ def _read_dataset( request=DataflowDatasetRequest( pipeline_ref=dataset.source_ref, revision=dataset.source_revision or 0, + run_ref=dataset.source_run_ref, parameters=source_parameters, row_limit=2_000, expected_definition_hash=dataset.definition_hash, diff --git a/src/govoplan_reporting/backend/manifest.py b/src/govoplan_reporting/backend/manifest.py index 6ce66e7..333aa9f 100644 --- a/src/govoplan_reporting/backend/manifest.py +++ b/src/govoplan_reporting/backend/manifest.py @@ -488,7 +488,7 @@ manifest = ModuleManifest( "schedules, exports, and publication providers replace unchecked SQL in the " "presentation layer. PostgreSQL executes bounded semantic plans when available. " "Signed drill contexts reauthorize contributor rows, and Files/Mail publication " - "adapters retain idempotent evidence. Dataflow and module read models remain source owners." + "adapters retain idempotent evidence. A dataset may pin one successful published Dataflow run, which is read from its exact Datasource materialization after both source boundaries reauthorize the current principal. Dataflow and module read models remain source owners." ), layer="available", documentation_types=("admin", "user"), diff --git a/src/govoplan_reporting/backend/schemas.py b/src/govoplan_reporting/backend/schemas.py index 3666e42..2479a32 100644 --- a/src/govoplan_reporting/backend/schemas.py +++ b/src/govoplan_reporting/backend/schemas.py @@ -101,6 +101,7 @@ class DatasetDefinition(BaseModel): source_kind: Literal["dataflow", "read_model", "static"] source_ref: str = Field(min_length=1, max_length=500) source_revision: int | None = Field(default=None, ge=1) + source_run_ref: str | None = Field(default=None, min_length=1, max_length=500) definition_hash: str | None = Field(default=None, min_length=1, max_length=128) source_parameters: dict[str, Any] = Field(default_factory=dict) static_rows: list[dict[str, Any]] = Field(default_factory=list, max_length=2_000) @@ -130,6 +131,8 @@ class DatasetDefinition(BaseModel): def validate_source_pin(self) -> "DatasetDefinition": if self.source_kind == "dataflow" and self.source_revision is None: raise ValueError("Dataflow datasets require a pinned source revision.") + if self.source_run_ref is not None and self.source_kind != "dataflow": + raise ValueError("Only Dataflow datasets can pin a source run.") if self.source_kind == "static" and not self.static_rows: raise ValueError("Static analytical datasets require static_rows.") names = [item.name for item in self.fields] diff --git a/tests/test_reporting_service.py b/tests/test_reporting_service.py index 83f8599..fb8ea06 100644 --- a/tests/test_reporting_service.py +++ b/tests/test_reporting_service.py @@ -8,6 +8,10 @@ from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.db.base import Base +from govoplan_core.core.dataflows import ( + CAPABILITY_DATAFLOW_DATASET_OUTPUT, + DataflowDatasetResult, +) from govoplan_core.core.files import ( CAPABILITY_FILES_ARTIFACT_STORE, ManagedArtifactRef, @@ -132,6 +136,34 @@ class ArtifactStore: ) +class DataflowOutput: + def __init__(self, rows: list[dict[str, object]]) -> None: + self.rows = tuple(dict(item) for item in rows) + self.last_request = None + + def list_outputs(self, *_args, **_kwargs): + return () + + def read_output(self, _session, _principal, *, request): + self.last_request = request + return DataflowDatasetResult( + pipeline_ref=request.pipeline_ref, + revision=request.revision, + definition_hash=request.expected_definition_hash or "pipeline-hash", + rows=self.rows, + total_rows=len(self.rows), + truncated=False, + output_hash="d" * 64, + executor_version="duckdb-v1", + run_ref=request.run_ref, + source_fingerprints=( + {"node_id": "source", "fingerprint": "source-v1"}, + ), + generated_at=NOW, + provenance={"immutable_run": bool(request.run_ref)}, + ) + + class ReportingServiceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") @@ -352,6 +384,44 @@ class ReportingServiceTests(unittest.TestCase): ) self.assertTrue(raised.exception.execution_id) + def test_dataflow_dataset_can_pin_an_exact_published_run(self) -> None: + payload = dataset_payload() + rows = list(payload.pop("static_rows")) + payload.update( + { + "source_kind": "dataflow", + "source_ref": "pipeline:monthly-comparison", + "source_revision": 4, + "source_run_ref": "dataflow-run:published-july", + "definition_hash": "pipeline-hash", + } + ) + self._create("dataset", "dataset-1", payload) + self._create("semantic_model", "semantic-1", semantic_payload()) + self._create("report", "report-1", report_payload()) + provider = DataflowOutput(rows) + registry = CapabilityRegistry() + registry.providers[CAPABILITY_DATAFLOW_DATASET_OUTPUT] = provider + + result = execute_report( + self.session, + self.principal, + registry=registry, + report_id="report-1", + report_revision=1, + parameters={}, + query=None, + idempotency_key="published-dataflow-run", + ) + + self.assertEqual("succeeded", result["status"]) + self.assertIsNotNone(provider.last_request) + self.assertEqual( + "dataflow-run:published-july", + provider.last_request.run_ref, + ) + self.assertTrue(result["provenance"]["source"]["immutable_run"]) + def test_restricted_access_and_service_scope_guards(self) -> None: self._create_report_graph( report_access={