GovOPlaN/govoplan-dataflow
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases Wiki Activity

Add governed Dataflow audience outputs

Browse Source
This commit is contained in:
zemion
2026-07-31 21:00:02 +02:00
parent 8724591bf9
commit cef2ba9482
12 changed files with 1226 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fixtures/golden/adrema-audience-selection/expected-output.json
+70
View File
@@ -0,0 +1,70 @@
[
{
"recipient_key": "identity:alice",
"display_name": "Alice Example",
"identity_id": "identity-alice",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"email": "alice@example.test",
"postal_address": null,
"account_id": null,
"portal_target": null,
"contact_point_id": "contact-email-alice",
"selected_channel": "email",
"distribution_status": "usable",
"exclusion_reason": null,
"policy_decision": "preference.explicit_email",
"_reconciliation_status": "match"
},
{
"recipient_key": "identity:bob",
"display_name": "Bob Example",
"identity_id": "identity-bob",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"email": null,
"postal_address": "Example Street 2, 10000 Exampletown",
"account_id": null,
"portal_target": null,
"contact_point_id": "contact-postal-bob",
"selected_channel": "postal",
"distribution_status": "suppressed",
"exclusion_reason": "preference.suppressed",
"policy_decision": "preference.suppressed",
"_reconciliation_status": "match"
},
{
"recipient_key": "identity:grace",
"display_name": "Grace Example",
"identity_id": "identity-grace",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"email": null,
"postal_address": "Example Street 3, 10000 Exampletown",
"account_id": null,
"portal_target": null,
"contact_point_id": "contact-postal-grace",
"selected_channel": "postal",
"distribution_status": "usable",
"exclusion_reason": null,
"policy_decision": "preference.default_postal",
"_reconciliation_status": "match"
},
{
"recipient_key": "identity:frank",
"display_name": "Frank Example",
"identity_id": "identity-frank",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"email": null,
"postal_address": null,
"account_id": null,
"portal_target": "portal:frank",
"contact_point_id": "contact-portal-frank",
"selected_channel": "portal",
"distribution_status": "usable",
"exclusion_reason": null,
"policy_decision": "manual.include_with_reason",
"_reconciliation_status": "match"
}
]
fixtures/golden/adrema-audience-selection/graph.json
+299
View File
@@ -0,0 +1,299 @@
{
"schema_version": 1,
"nodes": [
{
"id": "directory",
"type": "source.reference",
"label": "Frozen directory and contact snapshot",
"position": {"x": 40, "y": 60},
"config": {
"source_ref": "snapshot:directory-2026-01",
"source_name": "directory_snapshot",
"expected_fingerprint": "sha256:synthetic-directory-2026-01",
"consistency": "frozen",
"source_columns": [
"recipient_key", "display_name", "identity_id", "group_key",
"organization_unit_id", "function_id", "region", "purpose",
"effective", "effective_from", "effective_until",
"preferred_target", "preferred_contact_point_id", "preferred_channel",
"fallback_target", "fallback_contact_point_id", "fallback_channel",
"preference_decision", "sort_order"
]
}
},
{
"id": "group-filter",
"type": "filter",
"label": "Typed group",
"position": {"x": 260, "y": 60},
"config": {"column": "group_key", "operator": "eq", "value": {"$parameter": "group_key"}}
},
{
"id": "organization-filter",
"type": "filter",
"label": "Organization unit",
"position": {"x": 480, "y": 60},
"config": {"column": "organization_unit_id", "operator": "eq", "value": "district-north"}
},
{
"id": "function-filter",
"type": "filter",
"label": "Effective function",
"position": {"x": 700, "y": 60},
"config": {"column": "function_id", "operator": "eq", "value": "resident-service"}
},
{
"id": "region-filter",
"type": "filter",
"label": "Geography",
"position": {"x": 920, "y": 60},
"config": {"column": "region", "operator": "eq", "value": {"$parameter": "region"}}
},
{
"id": "purpose-filter",
"type": "filter",
"label": "Distribution purpose",
"position": {"x": 1140, "y": 60},
"config": {"column": "purpose", "operator": "eq", "value": {"$parameter": "purpose"}}
},
{
"id": "effective-filter",
"type": "filter",
"label": "Effective-dated assignment",
"position": {"x": 1360, "y": 60},
"config": {"column": "effective", "operator": "eq", "value": true}
},
{
"id": "contact-fallback",
"type": "calculate",
"label": "Resolve preferred or fallback contact",
"position": {"x": 1580, "y": 60},
"config": {
"calculations": [
{"target_column": "selected_target", "expression": "coalesce(preferred_target, fallback_target)", "result_type": "string"},
{"target_column": "selected_contact_point_id", "expression": "coalesce(preferred_contact_point_id, fallback_contact_point_id)", "result_type": "string"},
{"target_column": "selected_channel", "expression": "coalesce(preferred_channel, fallback_channel)", "result_type": "string"},
{"target_column": "email", "expression": "CASE WHEN selected_channel = 'email' THEN selected_target ELSE NULL END", "result_type": "string"},
{"target_column": "postal_address", "expression": "CASE WHEN selected_channel = 'postal' THEN selected_target ELSE NULL END", "result_type": "string"},
{"target_column": "account_id", "expression": "CASE WHEN selected_channel = 'internal_mail' THEN selected_target ELSE NULL END", "result_type": "string"},
{"target_column": "portal_target", "expression": "CASE WHEN selected_channel = 'portal' THEN selected_target ELSE NULL END", "result_type": "string"}
]
}
},
{
"id": "directory-fields",
"type": "select",
"label": "Canonical recipient candidate",
"position": {"x": 1800, "y": 60},
"config": {
"fields": [
"recipient_key", "display_name", "identity_id", "group_key",
"organization_unit_id", "function_id", "selected_target",
"selected_contact_point_id", "selected_channel", "email",
"postal_address", "account_id", "portal_target",
"preference_decision", "sort_order"
]
}
},
{
"id": "candidate-quality",
"type": "quality.rules",
"label": "Recipient quality gate",
"position": {"x": 2020, "y": 60},
"config": {
"rules": [
{"id": "recipient-key", "column": "recipient_key", "operator": "not_null"},
{"id": "delivery-target", "column": "selected_target", "operator": "not_null"},
{"id": "contact-point", "column": "selected_contact_point_id", "operator": "not_null"}
],
"action": "drop"
}
},
{
"id": "manual",
"type": "source.reference",
"label": "Frozen manual includes",
"position": {"x": 1580, "y": 300},
"config": {
"source_ref": "snapshot:manual-includes-2026-01",
"source_name": "manual_includes",
"expected_fingerprint": "sha256:synthetic-manual-includes-2026-01",
"consistency": "frozen",
"source_columns": [
"recipient_key", "display_name", "identity_id", "group_key",
"organization_unit_id", "function_id", "selected_target",
"selected_contact_point_id", "selected_channel", "email",
"postal_address", "account_id", "portal_target",
"preference_decision", "sort_order"
]
}
},
{
"id": "append-includes",
"type": "combine.union",
"label": "Append manual includes",
"position": {"x": 2240, "y": 170},
"config": {"mode": "all"}
},
{
"id": "deduplicate",
"type": "distinct",
"label": "One candidate per identity",
"position": {"x": 2460, "y": 170},
"config": {"columns": ["recipient_key"]}
},
{
"id": "suppressions",
"type": "source.reference",
"label": "Frozen channel suppressions",
"position": {"x": 2240, "y": 410},
"config": {
"source_ref": "snapshot:suppressions-2026-01",
"source_name": "channel_suppressions",
"expected_fingerprint": "sha256:synthetic-suppressions-2026-01",
"consistency": "frozen",
"source_columns": ["recipient_key", "suppression_reason"]
}
},
{
"id": "apply-suppressions",
"type": "combine.join",
"label": "Attach suppression decisions",
"position": {"x": 2680, "y": 250},
"config": {
"join_type": "left",
"left_keys": ["recipient_key"],
"right_keys": ["recipient_key"],
"right_prefix": "suppression_"
}
},
{
"id": "decision-columns",
"type": "calculate",
"label": "Classify channel decision",
"position": {"x": 2900, "y": 250},
"config": {
"calculations": [
{"target_column": "distribution_status", "expression": "CASE WHEN suppression_suppression_reason IS NOT NULL THEN 'suppressed' ELSE 'usable' END", "result_type": "string"},
{"target_column": "exclusion_reason", "expression": "suppression_suppression_reason", "result_type": "string"},
{"target_column": "policy_decision", "expression": "coalesce(suppression_suppression_reason, preference_decision)", "result_type": "string"}
]
}
},
{
"id": "decision-fields",
"type": "select",
"label": "Canonical audience decision",
"position": {"x": 3120, "y": 250},
"config": {
"fields": [
"recipient_key", "display_name", "identity_id",
"organization_unit_id", "function_id", "email", "postal_address",
"account_id", "portal_target",
{"column": "selected_contact_point_id", "alias": "contact_point_id"},
"selected_channel", "distribution_status", "exclusion_reason",
"policy_decision", "sort_order"
]
}
},
{
"id": "stable-order",
"type": "sort",
"label": "Stable recipient order",
"position": {"x": 3340, "y": 250},
"config": {
"fields": [
{"column": "sort_order", "direction": "asc"},
{"column": "recipient_key", "direction": "asc"}
]
}
},
{
"id": "expected",
"type": "source.reference",
"label": "Frozen expected audience",
"position": {"x": 3340, "y": 500},
"config": {
"source_ref": "snapshot:expected-audience-2026-01",
"source_name": "expected_audience",
"expected_fingerprint": "sha256:synthetic-expected-audience-2026-01",
"consistency": "frozen",
"source_columns": [
"recipient_key", "display_name", "identity_id",
"organization_unit_id", "function_id", "email", "postal_address",
"account_id", "portal_target", "contact_point_id",
"selected_channel", "distribution_status", "exclusion_reason",
"policy_decision"
]
}
},
{
"id": "reconcile",
"type": "reconcile.compare",
"label": "Reconcile expected audience",
"position": {"x": 3560, "y": 350},
"config": {
"left_keys": ["recipient_key"],
"right_keys": ["recipient_key"],
"compare_columns": [
"selected_channel", "contact_point_id", "distribution_status",
"exclusion_reason", "policy_decision"
],
"right_prefix": "actual_"
}
},
{
"id": "reconciled-only",
"type": "filter",
"label": "Require exact reconciliation",
"position": {"x": 3780, "y": 350},
"config": {"column": "_reconciliation_status", "operator": "eq", "value": "match"}
},
{
"id": "output-fields",
"type": "select",
"label": "Distribution-list dataset",
"position": {"x": 4000, "y": 350},
"config": {
"fields": [
"recipient_key", "display_name", "identity_id",
"organization_unit_id", "function_id", "email", "postal_address",
"account_id", "portal_target", "contact_point_id",
"selected_channel", "distribution_status", "exclusion_reason",
"policy_decision", "_reconciliation_status"
]
}
},
{
"id": "output",
"type": "output",
"label": "Pinned audience output",
"position": {"x": 4220, "y": 350},
"config": {}
}
],
"edges": [
{"id": "e01", "source": "directory", "target": "group-filter"},
{"id": "e02", "source": "group-filter", "target": "organization-filter"},
{"id": "e03", "source": "organization-filter", "target": "function-filter"},
{"id": "e04", "source": "function-filter", "target": "region-filter"},
{"id": "e05", "source": "region-filter", "target": "purpose-filter"},
{"id": "e06", "source": "purpose-filter", "target": "effective-filter"},
{"id": "e07", "source": "effective-filter", "target": "contact-fallback"},
{"id": "e08", "source": "contact-fallback", "target": "directory-fields"},
{"id": "e09", "source": "directory-fields", "target": "candidate-quality"},
{"id": "e10", "source": "candidate-quality", "target": "append-includes"},
{"id": "e11", "source": "manual", "target": "append-includes"},
{"id": "e12", "source": "append-includes", "target": "deduplicate"},
{"id": "e13", "source": "deduplicate", "target": "apply-suppressions", "target_port": "left"},
{"id": "e14", "source": "suppressions", "target": "apply-suppressions", "target_port": "right"},
{"id": "e15", "source": "apply-suppressions", "target": "decision-columns"},
{"id": "e16", "source": "decision-columns", "target": "decision-fields"},
{"id": "e17", "source": "decision-fields", "target": "stable-order"},
{"id": "e18", "source": "expected", "target": "reconcile", "target_port": "left"},
{"id": "e19", "source": "stable-order", "target": "reconcile", "target_port": "right"},
{"id": "e20", "source": "reconcile", "target": "reconciled-only"},
{"id": "e21", "source": "reconciled-only", "target": "output-fields"},
{"id": "e22", "source": "output-fields", "target": "output"}
]
}
fixtures/golden/adrema-audience-selection/inputs/directory-snapshot.json
+149
View File
@@ -0,0 +1,149 @@
[
{
"recipient_key": "identity:alice",
"display_name": "Alice Example",
"identity_id": "identity-alice",
"group_key": "residents",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"region": "north",
"purpose": "monthly-notice",
"effective": true,
"effective_from": "2025-01-01",
"effective_until": null,
"preferred_target": "alice@example.test",
"preferred_contact_point_id": "contact-email-alice",
"preferred_channel": "email",
"fallback_target": "Example Street 1, 10000 Exampletown",
"fallback_contact_point_id": "contact-postal-alice",
"fallback_channel": "postal",
"preference_decision": "preference.explicit_email",
"sort_order": 10
},
{
"recipient_key": "identity:bob",
"display_name": "Bob Example",
"identity_id": "identity-bob",
"group_key": "residents",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"region": "north",
"purpose": "monthly-notice",
"effective": true,
"effective_from": "2025-06-01",
"effective_until": "2027-01-01",
"preferred_target": null,
"preferred_contact_point_id": null,
"preferred_channel": null,
"fallback_target": "Example Street 2, 10000 Exampletown",
"fallback_contact_point_id": "contact-postal-bob",
"fallback_channel": "postal",
"preference_decision": "preference.default_postal",
"sort_order": 20
},
{
"recipient_key": "identity:grace",
"display_name": "Grace Example",
"identity_id": "identity-grace",
"group_key": "residents",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"region": "north",
"purpose": "monthly-notice",
"effective": true,
"effective_from": "2024-01-01",
"effective_until": null,
"preferred_target": null,
"preferred_contact_point_id": null,
"preferred_channel": null,
"fallback_target": "Example Street 3, 10000 Exampletown",
"fallback_contact_point_id": "contact-postal-grace",
"fallback_channel": "postal",
"preference_decision": "preference.default_postal",
"sort_order": 30
},
{
"recipient_key": "identity:carol",
"display_name": "Carol Example",
"identity_id": "identity-carol",
"group_key": "employees",
"organization_unit_id": "district-south",
"function_id": "finance-officer",
"region": "south",
"purpose": "monthly-notice",
"effective": true,
"effective_from": "2025-01-01",
"effective_until": null,
"preferred_target": "carol@example.test",
"preferred_contact_point_id": "contact-email-carol",
"preferred_channel": "email",
"fallback_target": null,
"fallback_contact_point_id": null,
"fallback_channel": null,
"preference_decision": "preference.explicit_email",
"sort_order": 40
},
{
"recipient_key": "identity:dana",
"display_name": "Dana Example",
"identity_id": "identity-dana",
"group_key": "residents",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"region": "north",
"purpose": "newsletter",
"effective": true,
"effective_from": "2025-01-01",
"effective_until": null,
"preferred_target": "dana@example.test",
"preferred_contact_point_id": "contact-email-dana",
"preferred_channel": "email",
"fallback_target": null,
"fallback_contact_point_id": null,
"fallback_channel": null,
"preference_decision": "preference.explicit_email",
"sort_order": 50
},
{
"recipient_key": "identity:erin",
"display_name": "Erin Example",
"identity_id": "identity-erin",
"group_key": "residents",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"region": "north",
"purpose": "monthly-notice",
"effective": false,
"effective_from": "2024-01-01",
"effective_until": "2025-01-01",
"preferred_target": "erin@example.test",
"preferred_contact_point_id": "contact-email-erin",
"preferred_channel": "email",
"fallback_target": null,
"fallback_contact_point_id": null,
"fallback_channel": null,
"preference_decision": "preference.explicit_email",
"sort_order": 60
},
{
"recipient_key": null,
"display_name": "Malformed Synthetic Row",
"identity_id": null,
"group_key": "residents",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"region": "north",
"purpose": "monthly-notice",
"effective": true,
"effective_from": "2025-01-01",
"effective_until": null,
"preferred_target": null,
"preferred_contact_point_id": null,
"preferred_channel": null,
"fallback_target": null,
"fallback_contact_point_id": null,
"fallback_channel": null,
"preference_decision": "preference.unresolved",
"sort_order": 70
}
]
fixtures/golden/adrema-audience-selection/inputs/expected-audience.json
+66
View File
@@ -0,0 +1,66 @@
[
{
"recipient_key": "identity:alice",
"display_name": "Alice Example",
"identity_id": "identity-alice",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"email": "alice@example.test",
"postal_address": null,
"account_id": null,
"portal_target": null,
"contact_point_id": "contact-email-alice",
"selected_channel": "email",
"distribution_status": "usable",
"exclusion_reason": null,
"policy_decision": "preference.explicit_email"
},
{
"recipient_key": "identity:bob",
"display_name": "Bob Example",
"identity_id": "identity-bob",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"email": null,
"postal_address": "Example Street 2, 10000 Exampletown",
"account_id": null,
"portal_target": null,
"contact_point_id": "contact-postal-bob",
"selected_channel": "postal",
"distribution_status": "suppressed",
"exclusion_reason": "preference.suppressed",
"policy_decision": "preference.suppressed"
},
{
"recipient_key": "identity:grace",
"display_name": "Grace Example",
"identity_id": "identity-grace",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"email": null,
"postal_address": "Example Street 3, 10000 Exampletown",
"account_id": null,
"portal_target": null,
"contact_point_id": "contact-postal-grace",
"selected_channel": "postal",
"distribution_status": "usable",
"exclusion_reason": null,
"policy_decision": "preference.default_postal"
},
{
"recipient_key": "identity:frank",
"display_name": "Frank Example",
"identity_id": "identity-frank",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"email": null,
"postal_address": null,
"account_id": null,
"portal_target": "portal:frank",
"contact_point_id": "contact-portal-frank",
"selected_channel": "portal",
"distribution_status": "usable",
"exclusion_reason": null,
"policy_decision": "manual.include_with_reason"
}
]
fixtures/golden/adrema-audience-selection/inputs/manual-includes.json
+36
View File
@@ -0,0 +1,36 @@
[
{
"recipient_key": "identity:alice",
"display_name": "Alice Example (duplicate manual include)",
"identity_id": "identity-alice",
"group_key": "residents",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"selected_target": "alice@example.test",
"selected_contact_point_id": "contact-email-alice",
"selected_channel": "email",
"email": "alice@example.test",
"postal_address": null,
"account_id": null,
"portal_target": null,
"preference_decision": "manual.include_duplicate",
"sort_order": 11
},
{
"recipient_key": "identity:frank",
"display_name": "Frank Example",
"identity_id": "identity-frank",
"group_key": "manual",
"organization_unit_id": "district-north",
"function_id": "resident-service",
"selected_target": "portal:frank",
"selected_contact_point_id": "contact-portal-frank",
"selected_channel": "portal",
"email": null,
"postal_address": null,
"account_id": null,
"portal_target": "portal:frank",
"preference_decision": "manual.include_with_reason",
"sort_order": 40
}
]
fixtures/golden/adrema-audience-selection/inputs/suppressions.json
+6
View File
@@ -0,0 +1,6 @@
[
{
"recipient_key": "identity:bob",
"suppression_reason": "preference.suppressed"
}
]
fixtures/golden/adrema-audience-selection/manifest.json
+92
View File
@@ -0,0 +1,92 @@
{
"id": "adrema-audience-selection",
"title": "Synthetic governed AdreMa audience selection",
"issue": "GovOPlaN/govoplan-dataflow#18",
"tenant_id": "tenant-synthetic-adrema",
"graph": "graph.json",
"expected_output": "expected-output.json",
"parameters": {
"group_key": "residents",
"region": "north",
"purpose": "monthly-notice"
},
"sources": {
"snapshot:directory-2026-01": {
"fixture": "inputs/directory-snapshot.json",
"tenant_id": "tenant-synthetic-adrema",
"provider": "connectors.synthetic-directory",
"fingerprint": "sha256:synthetic-directory-2026-01"
},
"snapshot:manual-includes-2026-01": {
"fixture": "inputs/manual-includes.json",
"tenant_id": "tenant-synthetic-adrema",
"provider": "datasources.static",
"fingerprint": "sha256:synthetic-manual-includes-2026-01"
},
"snapshot:suppressions-2026-01": {
"fixture": "inputs/suppressions.json",
"tenant_id": "tenant-synthetic-adrema",
"provider": "addresses.synthetic-preferences",
"fingerprint": "sha256:synthetic-suppressions-2026-01"
},
"snapshot:expected-audience-2026-01": {
"fixture": "inputs/expected-audience.json",
"tenant_id": "tenant-synthetic-adrema",
"provider": "reporting.synthetic-baseline",
"fingerprint": "sha256:synthetic-expected-audience-2026-01"
}
},
"expected_definition_hash": "2fec104dfda522dbc708ce61b237c062919703d62d2912ba621d5873ee5b975e",
"expected_execution_graph_hash": "d1cbb51c837810300044ed5f660e295ace49411265108b102731174d5587a85d",
"expected_output_hash": "f2d744354279f80393c6f894f7854569aa686f8292e6256fe429569d4410a72e",
"expected_source_fingerprints": [
{
"node_id": "directory",
"source_ref": "snapshot:directory-2026-01",
"source_name": "directory_snapshot",
"kind": "datasource",
"provider": "connectors.synthetic-directory",
"fingerprint": "sha256:synthetic-directory-2026-01",
"row_count": 7,
"preview_rows": 7,
"truncated": false
},
{
"node_id": "manual",
"source_ref": "snapshot:manual-includes-2026-01",
"source_name": "manual_includes",
"kind": "datasource",
"provider": "datasources.static",
"fingerprint": "sha256:synthetic-manual-includes-2026-01",
"row_count": 2,
"preview_rows": 2,
"truncated": false
},
{
"node_id": "suppressions",
"source_ref": "snapshot:suppressions-2026-01",
"source_name": "channel_suppressions",
"kind": "datasource",
"provider": "addresses.synthetic-preferences",
"fingerprint": "sha256:synthetic-suppressions-2026-01",
"row_count": 1,
"preview_rows": 1,
"truncated": false
},
{
"node_id": "expected",
"source_ref": "snapshot:expected-audience-2026-01",
"source_name": "expected_audience",
"kind": "datasource",
"provider": "reporting.synthetic-baseline",
"fingerprint": "sha256:synthetic-expected-audience-2026-01",
"row_count": 4,
"preview_rows": 4,
"truncated": false
}
],
"expected_recipients": ["identity:alice", "identity:grace", "identity:frank"],
"expected_exclusions": {
"identity:bob": "preference.suppressed"
}
}
src/govoplan_dataflow/backend/dataset_output.py
+200
View File
@@ -0,0 +1,200 @@
from __future__ import annotations
import hashlib
import json
from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.dataflows import (
DataflowDatasetDescriptor,
DataflowDatasetRequest,
DataflowDatasetResult,
DataflowRunConflictError,
DataflowRunUnavailableError,
)
from govoplan_core.security.time import utc_now
from govoplan_dataflow.backend.backends.base import ExecutionBudget
from govoplan_dataflow.backend.executor import PipelineExecutionError
from govoplan_dataflow.backend.governance import require_definition_action
from govoplan_dataflow.backend.schemas import PipelineGraph
from govoplan_dataflow.backend.service import (
_execute_pipeline_preview,
get_pipeline,
get_pipeline_revision,
list_pipelines,
)
from govoplan_dataflow.backend.subflows import substitute_parameters
class SqlDataflowDatasetOutputProvider:
def __init__(self, registry: object | None = None) -> None:
self.registry = registry
def list_outputs(
self,
session: object,
principal: object,
*,
query: str = "",
limit: int = 100,
) -> tuple[DataflowDatasetDescriptor, ...]:
typed_session, typed_principal = _contracts(session, principal)
normalized_query = query.strip().casefold()
result: list[DataflowDatasetDescriptor] = []
for pipeline in list_pipelines(
typed_session,
tenant_id=typed_principal.tenant_id,
):
if normalized_query and normalized_query not in pipeline.name.casefold():
continue
try:
decision = require_definition_action(
pipeline,
principal=typed_principal,
registry=self.registry,
action="run",
)
except PermissionError:
continue
revision = get_pipeline_revision(typed_session, pipeline=pipeline)
result.append(
DataflowDatasetDescriptor(
pipeline_ref=pipeline.id,
name=pipeline.name,
description=pipeline.description,
revision=revision.revision,
definition_hash=revision.content_hash,
status=pipeline.status,
updated_at=pipeline.updated_at,
parameters=dict(pipeline.metadata_.get("parameters") or {}),
provenance={
"module": "dataflow",
"scope_type": pipeline.scope_type,
"scope_id": pipeline.scope_id,
"policy_decision": decision.to_dict(),
},
)
)
if len(result) >= max(1, min(limit, 200)):
break
return tuple(result)
def read_output(
self,
session: object,
principal: object,
*,
request: DataflowDatasetRequest,
) -> DataflowDatasetResult:
typed_session, typed_principal = _contracts(session, principal)
row_limit = max(1, min(request.row_limit, 2_000))
pipeline = get_pipeline(
typed_session,
tenant_id=typed_principal.tenant_id,
pipeline_id=request.pipeline_ref,
)
decision = require_definition_action(
pipeline,
principal=typed_principal,
registry=self.registry,
action="run",
)
revision = get_pipeline_revision(
typed_session,
pipeline=pipeline,
revision=request.revision,
)
if (
request.expected_definition_hash
and request.expected_definition_hash != revision.content_hash
):
raise DataflowRunConflictError(
"The pinned Dataflow definition hash no longer matches the requested revision."
)
graph = PipelineGraph.model_validate(
substitute_parameters(revision.graph, dict(request.parameters))
)
try:
result, executor_version = _execute_pipeline_preview(
graph,
session=typed_session,
principal=typed_principal,
registry=self.registry,
backend="auto",
row_limit=row_limit,
preview_node_id=None,
budget=ExecutionBudget(
max_output_rows=row_limit,
max_batch_bytes=4_000_000,
max_wall_seconds=5.0,
),
)
except PipelineExecutionError as exc:
raise DataflowRunUnavailableError(
f"The pinned Dataflow output could not be evaluated: {exc}"
) from exc
source_fingerprints = tuple(
dict(item) for item in result.source_fingerprints
)
if request.expected_source_fingerprints and not _fingerprints_match(
request.expected_source_fingerprints,
source_fingerprints,
):
raise DataflowRunConflictError(
"Dataflow source fingerprints differ from the pinned source state."
)
output_hash = hashlib.sha256(
json.dumps(
result.rows,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=True,
default=str,
).encode("utf-8")
).hexdigest()
return DataflowDatasetResult(
pipeline_ref=pipeline.id,
revision=revision.revision,
definition_hash=revision.content_hash,
rows=tuple(dict(item) for item in result.rows),
total_rows=result.total_rows,
truncated=result.truncated,
output_hash=output_hash,
executor_version=executor_version,
source_fingerprints=source_fingerprints,
diagnostics=tuple(
item.model_dump(mode="json") for item in result.diagnostics
),
generated_at=utc_now(),
provenance={
"module": "dataflow",
"scope_type": pipeline.scope_type,
"scope_id": pipeline.scope_id,
"policy_decision": decision.to_dict(),
"parameters": dict(request.parameters),
},
)
def dataset_output_provider(context: object | None = None):
return SqlDataflowDatasetOutputProvider(getattr(context, "registry", None))
def _fingerprints_match(expected, actual) -> bool:
def normalized(values):
return sorted(
json.dumps(dict(item), sort_keys=True, separators=(",", ":"), default=str)
for item in values
)
return normalized(expected) == normalized(actual)
def _contracts(session: object, principal: object):
if not hasattr(session, "scalar") or not hasattr(session, "scalars"):
raise TypeError("Dataflow dataset output requires a SQLAlchemy session.")
if not isinstance(principal, ApiPrincipal):
raise TypeError("Dataflow dataset output requires an API principal.")
return session, principal
__all__ = ["SqlDataflowDatasetOutputProvider", "dataset_output_provider"]
src/govoplan_dataflow/backend/manifest.py
+6 -1
View File
@@ -11,6 +11,7 @@ from govoplan_core.core.access import (
CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
)
from govoplan_core.core.dataflows import (
CAPABILITY_DATAFLOW_DATASET_OUTPUT,
CAPABILITY_DATAFLOW_RUN_LIFECYCLE,
CAPABILITY_DATAFLOW_RUN_WORKER,
CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
@@ -279,7 +280,7 @@ manifest = ModuleManifest(
ModuleInterfaceProvider(name="dataflow.pipeline_preview", version=MODULE_VERSION),
ModuleInterfaceProvider(name="dataflow.run_lifecycle", version=MODULE_VERSION),
ModuleInterfaceProvider(name="dataflow.run_worker", version=MODULE_VERSION),
ModuleInterfaceProvider(name="dataflow.dataset_output", version=MODULE_VERSION),
ModuleInterfaceProvider(name=CAPABILITY_DATAFLOW_DATASET_OUTPUT, version=MODULE_VERSION),
ModuleInterfaceProvider(
name="dataflow.trigger_dispatcher",
version=MODULE_VERSION,
@@ -378,6 +379,10 @@ manifest = ModuleManifest(
),
route_factory=_dataflow_router,
capability_factories={
CAPABILITY_DATAFLOW_DATASET_OUTPUT: lambda context: __import__(
"govoplan_dataflow.backend.dataset_output",
fromlist=["dataset_output_provider"],
).dataset_output_provider(context),
CAPABILITY_DATAFLOW_RUN_LIFECYCLE: _run_provider,
CAPABILITY_DATAFLOW_RUN_WORKER: _run_worker,
CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER: _trigger_provider,
tests/test_dataset_output.py
+150
View File
@@ -0,0 +1,150 @@
from __future__ import annotations
import unittest
from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.dataflows import DataflowDatasetRequest, DataflowRunConflictError
from govoplan_core.db.base import Base
from govoplan_core.db.session import configure_database, reset_database
from govoplan_dataflow.backend.dataset_output import SqlDataflowDatasetOutputProvider
from govoplan_dataflow.backend.db.models import DataflowPipeline, DataflowPipelineRevision
from govoplan_dataflow.backend.schemas import PipelineGraph
from govoplan_dataflow.backend.service import definition_hash
def principal(tenant_id: str = "tenant-1") -> ApiPrincipal:
return ApiPrincipal(
principal=PrincipalRef(
account_id="account-1",
membership_id="membership-1",
tenant_id=tenant_id,
scopes=frozenset({"dataflow:pipeline:read", "dataflow:pipeline:run"}),
),
account=object(),
user=object(),
)
class DataflowDatasetOutputTests(unittest.TestCase):
def setUp(self) -> None:
self.database = configure_database("sqlite:///:memory:")
Base.metadata.create_all(
self.database.engine,
tables=[DataflowPipeline.__table__, DataflowPipelineRevision.__table__],
)
def tearDown(self) -> None:
reset_database(dispose=True)
def test_pinned_revision_executes_through_typed_registry(self) -> None:
graph = PipelineGraph.model_validate(
{
"nodes": [
{
"id": "source",
"type": "source.inline",
"label": "Source",
"position": {"x": 0, "y": 0},
"config": {
"source_name": "audience",
"rows": [
{"recipient_key": "one", "email": "one@example.test"},
{"recipient_key": "two", "email": "two@example.test"},
]
},
},
{
"id": "output",
"type": "output",
"label": "Output",
"position": {"x": 200, "y": 0},
"config": {},
},
],
"edges": [{"id": "edge-1", "source": "source", "target": "output"}],
}
)
content_hash = definition_hash(graph, None)
with self.database.session() as session:
pipeline = DataflowPipeline(
id="pipeline-1",
tenant_id="tenant-1",
scope_type="tenant",
scope_id="tenant-1",
definition_kind="flow",
name="Audience",
status="active",
current_revision=1,
allow_run=True,
metadata_={},
)
revision = DataflowPipelineRevision(
id="revision-1",
tenant_id="tenant-1",
pipeline_id=pipeline.id,
revision=1,
graph=graph.model_dump(mode="json"),
sql_text=None,
editor_mode="graph",
content_hash=content_hash,
)
session.add_all((pipeline, revision))
session.flush()
provider = SqlDataflowDatasetOutputProvider()
descriptors = provider.list_outputs(session, principal())
result = provider.read_output(
session,
principal(),
request=DataflowDatasetRequest(
pipeline_ref=pipeline.id,
revision=1,
expected_definition_hash=content_hash,
),
)
self.assertEqual(1, len(descriptors))
self.assertEqual(2, result.total_rows)
self.assertEqual("one", result.rows[0]["recipient_key"])
self.assertEqual(64, len(result.output_hash))
replay = provider.read_output(
session,
principal(),
request=DataflowDatasetRequest(
pipeline_ref=pipeline.id,
revision=1,
expected_definition_hash=content_hash,
expected_source_fingerprints=result.source_fingerprints,
),
)
self.assertEqual(result.output_hash, replay.output_hash)
with self.assertRaises(DataflowRunConflictError):
provider.read_output(
session,
principal(),
request=DataflowDatasetRequest(
pipeline_ref=pipeline.id,
revision=1,
expected_definition_hash="wrong",
),
)
with self.assertRaises(DataflowRunConflictError):
provider.read_output(
session,
principal(),
request=DataflowDatasetRequest(
pipeline_ref=pipeline.id,
revision=1,
expected_definition_hash=content_hash,
expected_source_fingerprints=(
{"node_id": "source", "fingerprint": "changed"},
),
),
)
if __name__ == "__main__":
unittest.main()
tests/test_golden_flows.py
+147 -5
View File
@@ -1,12 +1,19 @@
from __future__ import annotations
import json
import hashlib
import unittest
from pathlib import Path
from govoplan_dataflow.backend.executor import execute_preview
from govoplan_dataflow.backend.graph import validate_graph
from govoplan_dataflow.backend.executor import (
PipelineExecutionError,
ResolvedSource,
execute_preview,
)
from govoplan_dataflow.backend.expressions import ExpressionError, parse_expression
from govoplan_dataflow.backend.graph import definition_hash, validate_graph
from govoplan_dataflow.backend.schemas import PipelineGraph
from govoplan_dataflow.backend.subflows import substitute_parameters
FIXTURES = Path(__file__).parents[1] / "fixtures" / "golden"
@@ -21,13 +28,18 @@ class GoldenFlowTests(unittest.TestCase):
for directory in directories:
with self.subTest(flow=directory.name):
manifest = _json(directory / "manifest.json")
graph_payload = _json(directory / str(manifest["graph"]))
raw_graph_payload = _json(directory / str(manifest["graph"]))
graph_payload = json.loads(json.dumps(raw_graph_payload))
for item in graph_payload["nodes"]:
fixture = item.get("config", {}).get("fixture")
if fixture:
if fixture and item["type"] == "source.inline":
item["config"]["rows"] = _json(
directory / "inputs" / fixture
)
graph_payload = substitute_parameters(
graph_payload,
dict(manifest.get("parameters") or {}),
)
graph = PipelineGraph.model_validate(graph_payload)
diagnostics = validate_graph(graph)
self.assertEqual(
@@ -39,17 +51,147 @@ class GoldenFlowTests(unittest.TestCase):
],
)
result = execute_preview(graph, row_limit=500)
resolver = _source_resolver(directory, manifest)
result = execute_preview(
graph,
row_limit=500,
source_resolver=resolver,
)
self.assertEqual(
_json(directory / str(manifest["expected_output"])),
result.rows,
)
if "expected_definition_hash" in manifest:
self.assertEqual(
manifest["expected_definition_hash"],
definition_hash(PipelineGraph.model_validate(raw_graph_payload)),
)
if "expected_execution_graph_hash" in manifest:
self.assertEqual(
manifest["expected_execution_graph_hash"],
definition_hash(graph),
)
if "expected_output_hash" in manifest:
self.assertEqual(
manifest["expected_output_hash"],
_rows_hash(result.rows),
)
if "expected_source_fingerprints" in manifest:
self.assertEqual(
manifest["expected_source_fingerprints"],
result.source_fingerprints,
)
full_result = execute_preview(
graph,
row_limit=5_000,
source_resolver=resolver,
)
self.assertEqual(result.rows, full_result.rows)
self.assertFalse(full_result.truncated)
def test_adrema_fixture_rejects_untrusted_sources_and_unsafe_expression(self) -> None:
directory = FIXTURES / "adrema-audience-selection"
manifest = _json(directory / "manifest.json")
graph = PipelineGraph.model_validate(
substitute_parameters(
_json(directory / "graph.json"),
dict(manifest["parameters"]),
)
)
drifted_manifest = {
**manifest,
"sources": {
**manifest["sources"],
"snapshot:directory-2026-01": {
**manifest["sources"]["snapshot:directory-2026-01"],
"fingerprint": "sha256:changed-source",
},
},
}
with self.assertRaisesRegex(PipelineExecutionError, "fingerprint"):
execute_preview(
graph,
row_limit=500,
source_resolver=_source_resolver(directory, drifted_manifest),
)
cross_tenant_manifest = {
**manifest,
"sources": {
**manifest["sources"],
"snapshot:directory-2026-01": {
**manifest["sources"]["snapshot:directory-2026-01"],
"tenant_id": "tenant-other",
},
},
}
with self.assertRaisesRegex(PipelineExecutionError, "tenant"):
execute_preview(
graph,
row_limit=500,
source_resolver=_source_resolver(directory, cross_tenant_manifest),
)
unknown_source_graph = graph.model_copy(deep=True)
unknown_source_graph.nodes[0].config["source_ref"] = "connector:untrusted"
with self.assertRaisesRegex(PipelineExecutionError, "pinned provider set"):
execute_preview(
unknown_source_graph,
row_limit=500,
source_resolver=_source_resolver(directory, manifest),
)
with self.assertRaises(ExpressionError):
parse_expression("read_csv('/tmp/private.csv')")
def _json(path: Path):
return json.loads(path.read_text(encoding="utf-8"))
def _source_resolver(directory: Path, manifest: dict):
sources = dict(manifest.get("sources") or {})
tenant_id = str(manifest.get("tenant_id") or "")
def resolve(node, limit: int) -> ResolvedSource:
source_ref = str(node.config.get("source_ref") or "")
if source_ref not in sources:
raise ValueError(f"Fixture source {source_ref!r} is outside the pinned provider set.")
source = sources[source_ref]
if str(source.get("tenant_id") or "") != tenant_id:
raise ValueError(
f"Fixture source {source_ref!r} belongs to a different tenant."
)
fingerprint = str(source["fingerprint"])
expected = str(node.config.get("expected_fingerprint") or "")
if expected and expected != fingerprint:
raise ValueError(
f"Fixture source fingerprint changed for {source_ref!r}."
)
rows = _json(directory / str(source["fixture"]))
bounded = rows[:limit]
return ResolvedSource(
rows=tuple(dict(item) for item in bounded),
source_ref=source_ref,
provider=str(source["provider"]),
fingerprint=fingerprint,
total_rows=len(rows),
truncated=len(bounded) < len(rows),
)
return resolve
def _rows_hash(rows: object) -> str:
return hashlib.sha256(
json.dumps(
rows,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=True,
).encode("utf-8")
).hexdigest()
if __name__ == "__main__":
unittest.main()
tests/test_manifest.py
+5
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import unittest
from govoplan_core.core.dataflows import (
CAPABILITY_DATAFLOW_DATASET_OUTPUT,
CAPABILITY_DATAFLOW_RUN_LIFECYCLE,
CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
)
@@ -31,6 +32,10 @@ class DataflowManifestTests(unittest.TestCase):
CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
manifest.capability_factories,
)
self.assertIn(
CAPABILITY_DATAFLOW_DATASET_OUTPUT,
manifest.capability_factories,
)
self.assertEqual(
{
"dataflow.pipeline_catalog",