Files
govoplan-dataflow/src/govoplan_dataflow/backend/manifest.py
T

640 lines
23 KiB
Python

from __future__ import annotations
from pathlib import Path
from govoplan_core.core.module_guards import (
drop_table_retirement_provider,
persistent_table_uninstall_guard,
)
from govoplan_core.core.access import (
CAPABILITY_ACCESS_DIRECTORY,
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,
)
from govoplan_core.core.modules import (
DocumentationTopic,
FrontendModule,
FrontendRoute,
MigrationSpec,
ModuleInterfaceProvider,
ModuleInterfaceRequirement,
ModuleContext,
ModuleManifest,
NavItem,
PermissionDefinition,
RoleTemplate,
)
from govoplan_core.core.provider_governance import declared_module_architecture
from govoplan_core.core.datasources import (
CAPABILITY_DATASOURCE_CATALOGUE,
CAPABILITY_DATASOURCE_LIFECYCLE,
CAPABILITY_DATASOURCE_PUBLICATION,
)
from govoplan_core.core.policy import (
CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
)
from govoplan_core.core.notifications import (
CAPABILITY_NOTIFICATIONS_DISPATCH,
)
from govoplan_core.core.references import CAPABILITY_ACCESS_REFERENCE_OPTIONS
from govoplan_core.core.search import SearchSourceProviderRegistration
from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base
from govoplan_dataflow.backend.db import models as dataflow_models
MODULE_ID = "dataflow"
MODULE_NAME = "Dataflow"
MODULE_VERSION = "0.1.14"
READ_SCOPE = "dataflow:pipeline:read"
WRITE_SCOPE = "dataflow:pipeline:write"
RUN_SCOPE = "dataflow:pipeline:run"
ADMIN_SCOPE = "dataflow:pipeline:admin"
TRIGGER_READ_SCOPE = "dataflow:trigger:read"
TRIGGER_WRITE_SCOPE = "dataflow:trigger:write"
TRIGGER_DISPATCH_SCOPE = "dataflow:trigger:dispatch"
def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
module_id, resource, action = scope.split(":", 2)
return PermissionDefinition(
scope=scope,
label=label,
description=description,
category="Dataflow",
level="tenant",
module_id=module_id,
resource=resource,
action=action,
)
PERMISSIONS = (
_permission(
READ_SCOPE,
"View data pipelines",
"Read pipeline definitions, revisions, validation diagnostics, and run summaries.",
),
_permission(
WRITE_SCOPE,
"Manage data pipelines",
"Create, edit, version, archive, and remove data pipeline definitions.",
),
_permission(
RUN_SCOPE,
"Run data pipelines",
"Validate and execute bounded previews or approved pipeline runs.",
),
_permission(
ADMIN_SCOPE,
"Administer Dataflow",
"Manage every tenant pipeline and future execution, retention, and publication policies.",
),
_permission(
TRIGGER_READ_SCOPE,
"View Dataflow triggers",
"Inspect schedule and event trigger configuration, status, provenance, and deliveries.",
),
_permission(
TRIGGER_WRITE_SCOPE,
"Manage Dataflow triggers",
"Create, edit, enable, disable, and remove governed Dataflow triggers.",
),
_permission(
TRIGGER_DISPATCH_SCOPE,
"Dispatch Dataflow triggers",
"Ingest trusted events and dispatch queued automated Dataflow runs.",
),
)
ROLE_TEMPLATES = (
RoleTemplate(
slug="dataflow_designer",
name="Dataflow designer",
description="Design, validate, and preview governed data pipelines.",
permissions=(
READ_SCOPE,
WRITE_SCOPE,
RUN_SCOPE,
TRIGGER_READ_SCOPE,
TRIGGER_WRITE_SCOPE,
),
),
RoleTemplate(
slug="dataflow_operator",
name="Dataflow operator",
description="Inspect and run approved data pipelines without changing their definitions.",
permissions=(READ_SCOPE, RUN_SCOPE, TRIGGER_READ_SCOPE),
),
RoleTemplate(
slug="dataflow_viewer",
name="Dataflow viewer",
description="Inspect pipeline definitions, revisions, diagnostics, and run summaries.",
permissions=(READ_SCOPE, TRIGGER_READ_SCOPE),
),
)
DOCUMENTATION = (
DocumentationTopic(
id="dataflow.module-boundary",
title="Dataflow module boundary",
summary="Versioned tabular transformations with graphical and constrained SQL editing.",
body=(
"Dataflow owns canonical pipeline graphs, immutable revisions, validation, constrained "
"SQL compilation, preview and run diagnostics, and lineage references. Datasources owns "
"the governed catalogue and materializations, while Connectors owns external acquisition "
"and credentials; Reporting owns analytical presentation and exports; "
"Workflow owns orchestration and human handoffs; Risk Compliance owns sanctions review "
"semantics and policy gates. User SQL is compiled into approved transforms and is never "
"passed unchecked to a backing database."
),
layer="available",
documentation_types=("admin", "user"),
audience=("operator", "module_admin", "power_user", "product_owner"),
order=75,
related_modules=(
"datasources",
"connectors",
"files",
"reporting",
"workflow_engine",
"risk_compliance",
"notifications",
"policy",
"audit",
),
metadata={
"first_slice": (
"Inline and governed datasources, union, join, filter, deduplication, select, "
"typed expressions, conversion, quality and reconciliation, reusable subflows, "
"aggregate, sort, limit, output, revisioning, and bounded preview."
),
"sql_safety": "Constrained AST compilation only; no pass-through execution.",
"run_recovery": (
"Database-only runs commit atomically with Core recovery evidence. "
"Output publication uses forward recovery and blocks blind retry "
"when provider acknowledgement is uncertain."
),
"help_contexts": [
"dataflow.page",
"dataflow.library",
"dataflow.graph",
"dataflow.sql",
"dataflow.inspector",
"dataflow.results",
"dataflow.state.read-only",
],
},
),
DocumentationTopic(
id="dataflow.reference.nodes-and-expressions",
title="Dataflow nodes and expressions",
summary="Typed node inputs, expressions, schema propagation, and bounded intermediate previews.",
body=(
"Every graph node declares typed inputs, configuration, output schema, and validation rules. "
"Source nodes pin inline content or governed Datasource references; combine, filter, transform, "
"quality, reconciliation, reusable-subflow, and output nodes remain explicit in the canonical graph. "
"Expressions use the typed Dataflow expression language and never execute arbitrary host or database "
"code. Selecting a node may request a bounded intermediate preview; preview rows are transient, "
"privacy-filtered for the actor, and are not retained as run output. SQL editing compiles into the same "
"canonical graph, so unsupported statements are diagnostics rather than pass-through SQL."
),
layer="available",
documentation_types=("admin", "user"),
audience=("operator", "module_admin", "power_user", "data_steward"),
order=76,
related_modules=("datasources", "connectors", "policy", "audit"),
metadata={
"help_contexts": [
"dataflow.field.node-name",
"dataflow.field.source",
"dataflow.field.expression",
"dataflow.field.schema",
"dataflow.action.preview-node",
],
},
),
DocumentationTopic(
id="dataflow.reference.fields-and-consequences",
title="Dataflow fields and lifecycle consequences",
summary="Definition scope, revision, reuse, automation, execution, publication, promotion, and deletion semantics.",
body=(
"Scope determines ownership and Policy inheritance. Templates can be derived but not run; complete "
"flows may be previewed, revisioned, automated, and executed when effective Policy allows it. Saving "
"appends an immutable revision. A scoped copy pins its source revision and content hash. Triggers pin "
"the revision and authorization grant, then re-evaluate authority for every delivery. Runs create "
"durable command and recovery evidence. Publishing creates a governed Datasource materialization, and "
"environment promotion changes which immutable revision is eligible for staging or production runs. "
"Deletion prevents future use while retained run, deployment, lineage, audit, and recovery evidence "
"continues under its retention policy."
),
layer="available",
documentation_types=("admin", "user"),
audience=("operator", "module_admin", "power_user", "product_owner"),
order=77,
related_modules=("datasources", "workflow_engine", "notifications", "policy", "audit"),
metadata={
"help_contexts": [
"dataflow.field.scope",
"dataflow.field.definition-kind",
"dataflow.action.save",
"dataflow.action.derive",
"dataflow.action.trigger",
"dataflow.action.delete",
],
"consequence_classes": {
"save_revision": "Appends an immutable pipeline definition revision.",
"derive_copy": "Creates a separately governed copy pinned to the source revision and hash.",
"configure_trigger": "Creates or changes an automation command with revision and authorization evidence.",
"delete_pipeline": "Prevents future use while retained evidence remains governed.",
},
},
),
DocumentationTopic(
id="dataflow.execution-and-recovery",
title="Dataflow execution, publication, and recovery",
summary="Pinned runs, environment promotion, output publication, cancellation, reconciliation, and retained evidence.",
body=(
"Every run is pinned to an immutable revision and idempotency key. The queue records actor, authority, "
"environment, progress, cancellation, output, and recovery state. Database-only runs commit atomically. "
"Publication to a governed Datasource uses forward recovery: an unknown provider outcome is reconciled "
"before retry so output is not duplicated. Staging and production promotion is explicit and does not "
"rewrite a revision. Cancellation is best effort once external work has started; the final evidence "
"states whether work stopped, completed, failed, or requires operator reconciliation."
),
layer="available",
documentation_types=("admin", "user"),
audience=("operator", "module_admin", "power_user", "security_admin"),
order=78,
related_modules=("datasources", "notifications", "policy", "audit", "ops"),
metadata={
"help_contexts": [
"dataflow.runs",
"dataflow.action.queue-run",
"dataflow.action.publish",
"dataflow.action.promote-staging",
"dataflow.action.promote-production",
"dataflow.state.recovery-attention",
],
"consequence_classes": {
"queue_run": "Creates a durable asynchronous command and authorization evidence.",
"publish_output": "Creates or updates a governed Datasource and appends a materialization.",
"promote_revision": "Makes an immutable revision eligible in a higher execution environment.",
"cancel_run": "Requests cancellation; already acknowledged external effects may remain.",
},
},
),
)
def _dataflow_router(context: ModuleContext):
from govoplan_dataflow.backend.runtime import configure_runtime
configure_runtime(registry=context.registry, settings=context.settings)
from govoplan_dataflow.backend.router import router
return router
def _run_provider(context: ModuleContext):
from govoplan_dataflow.backend.service import SqlDataflowRunLifecycleProvider
return SqlDataflowRunLifecycleProvider(registry=context.registry)
def _run_worker(context: ModuleContext):
from govoplan_dataflow.backend.run_worker import SqlDataflowRunWorker
return SqlDataflowRunWorker(registry=context.registry)
def _trigger_provider(context: ModuleContext):
from govoplan_dataflow.backend.triggers import (
SqlDataflowTriggerDispatcher,
)
return SqlDataflowTriggerDispatcher(registry=context.registry)
def _search_source_provider(context: ModuleContext):
from govoplan_dataflow.backend.search_source import (
create_dataflow_search_source,
)
return create_dataflow_search_source(context)
def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
return {
"dataflow_pipelines": (
session.query(dataflow_models.DataflowPipeline)
.filter(
dataflow_models.DataflowPipeline.tenant_id == tenant_id,
dataflow_models.DataflowPipeline.deleted_at.is_(None),
)
.count()
),
"dataflow_runs": (
session.query(dataflow_models.DataflowRun)
.filter(dataflow_models.DataflowRun.tenant_id == tenant_id)
.count()
),
"dataflow_triggers": (
session.query(dataflow_models.DataflowTrigger)
.filter(dataflow_models.DataflowTrigger.tenant_id == tenant_id)
.count()
),
"dataflow_trigger_deliveries": (
session.query(dataflow_models.DataflowTriggerDelivery)
.filter(
dataflow_models.DataflowTriggerDelivery.tenant_id
== tenant_id
)
.count()
),
}
manifest = ModuleManifest(
id=MODULE_ID,
name=MODULE_NAME,
version=MODULE_VERSION,
dependencies=(),
optional_dependencies=(
"access",
"audit",
"datasources",
"files",
"notifications",
"policy",
"reporting",
"risk_compliance",
"search",
"workflow_engine",
),
optional_capabilities=(
CAPABILITY_ACCESS_DIRECTORY,
CAPABILITY_ACCESS_REFERENCE_OPTIONS,
CAPABILITY_DATASOURCE_CATALOGUE,
CAPABILITY_DATASOURCE_LIFECYCLE,
CAPABILITY_DATASOURCE_PUBLICATION,
CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
CAPABILITY_NOTIFICATIONS_DISPATCH,
CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
),
provides_interfaces=(
ModuleInterfaceProvider(name="dataflow.pipeline_catalog", version=MODULE_VERSION),
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=CAPABILITY_DATAFLOW_DATASET_OUTPUT, version=MODULE_VERSION),
ModuleInterfaceProvider(
name="dataflow.trigger_dispatcher",
version=MODULE_VERSION,
),
),
requires_interfaces=(
ModuleInterfaceRequirement(
name=CAPABILITY_ACCESS_REFERENCE_OPTIONS,
version_min="0.1.0",
version_max_exclusive="0.2.0",
optional=True,
),
ModuleInterfaceRequirement(
name="datasources.catalogue",
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,
),
ModuleInterfaceRequirement(
name="policy.definition_governance",
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,
),
ModuleInterfaceRequirement(
name="auth.automation_principal",
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,
),
ModuleInterfaceRequirement(
name="datasources.lifecycle",
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,
),
ModuleInterfaceRequirement(
name="datasources.publication",
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,
),
ModuleInterfaceRequirement(
name=CAPABILITY_NOTIFICATIONS_DISPATCH,
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,
),
ModuleInterfaceRequirement(
name="search.source",
version_min="1.0.0",
version_max_exclusive="2.0.0",
optional=True,
),
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
nav_items=(
NavItem(
path="/dataflow",
label="Dataflow",
icon="waypoints",
required_any=(READ_SCOPE, ADMIN_SCOPE),
order=72,
),
),
frontend=FrontendModule(
module_id=MODULE_ID,
package_name="@govoplan/dataflow-webui",
routes=(
FrontendRoute(
path="/dataflow",
component="DataflowPage",
required_any=(READ_SCOPE, ADMIN_SCOPE),
order=72,
),
),
nav_items=(
NavItem(
path="/dataflow",
label="Dataflow",
icon="waypoints",
required_any=(READ_SCOPE, ADMIN_SCOPE),
order=72,
),
),
view_surfaces=(
ViewSurface(
id="dataflow.page",
module_id=MODULE_ID,
kind="route",
label="Dataflow",
order=72,
),
ViewSurface(
id="dataflow.library",
module_id=MODULE_ID,
kind="section",
label="Pipeline library",
parent_id="dataflow.page",
order=10,
),
ViewSurface(
id="dataflow.graph",
module_id=MODULE_ID,
kind="section",
label="Graph editor",
parent_id="dataflow.page",
order=20,
),
ViewSurface(
id="dataflow.sql",
module_id=MODULE_ID,
kind="section",
label="Constrained SQL editor",
parent_id="dataflow.page",
order=30,
),
ViewSurface(
id="dataflow.inspector",
module_id=MODULE_ID,
kind="section",
label="Node inspector",
parent_id="dataflow.page",
order=40,
),
ViewSurface(
id="dataflow.results",
module_id=MODULE_ID,
kind="section",
label="Preview and diagnostics",
parent_id="dataflow.page",
order=50,
),
ViewSurface(
id="dataflow.triggers",
module_id=MODULE_ID,
kind="action",
label="Automation triggers",
parent_id="dataflow.page",
order=60,
),
ViewSurface(
id="dataflow.runs",
module_id=MODULE_ID,
kind="action",
label="Runs and deployments",
parent_id="dataflow.page",
order=70,
),
ViewSurface(
id="dataflow.widget.pipelines",
module_id=MODULE_ID,
kind="section",
label="Dataflows widget",
order=70,
),
),
),
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,
},
search_sources=(
SearchSourceProviderRegistration(
id="dataflow.pipelines",
factory=_search_source_provider,
),
),
tenant_summary_providers=(_tenant_summary,),
migration_spec=MigrationSpec(
module_id=MODULE_ID,
metadata=Base.metadata,
script_location=str(Path(__file__).with_name("migrations") / "versions"),
retirement_supported=True,
retirement_provider=drop_table_retirement_provider(
dataflow_models.DataflowTriggerDelivery,
dataflow_models.DataflowTrigger,
dataflow_models.DataflowRun,
dataflow_models.DataflowPipelineDeployment,
dataflow_models.DataflowPipelineRevision,
dataflow_models.DataflowPipeline,
label="Dataflow",
),
retirement_notes=(
"Destructive retirement drops Dataflow definitions, revisions, and run evidence "
"after the installer captures a database snapshot."
),
),
uninstall_guard_providers=(
persistent_table_uninstall_guard(
dataflow_models.DataflowPipeline,
dataflow_models.DataflowPipelineRevision,
dataflow_models.DataflowPipelineDeployment,
dataflow_models.DataflowRun,
dataflow_models.DataflowTrigger,
dataflow_models.DataflowTriggerDelivery,
label="Dataflow",
),
),
documentation=DOCUMENTATION,
architecture=declared_module_architecture(
layer="data_reporting_integration",
kind="runtime",
maturity="vertical_slice",
documentation_ref="README.md",
test_ref="tests/test_golden_flows.py",
known_limits=(
"Execution adapters do not yet cover every declared node family.",
),
owned_concepts=("dataflow definition", "dataflow revision", "dataflow run", "transformation graph"),
non_owned_concepts=("datasource binding", "connector transport", "report presentation", "workflow task"),
recovery_docs=("README.md", "docs/DURABLE_RUN_RECOVERY.md"),
security_docs=("README.md",),
operations_docs=("README.md",),
),
)
def get_manifest() -> ModuleManifest:
return manifest
__all__ = [
"ADMIN_SCOPE",
"MODULE_ID",
"MODULE_NAME",
"MODULE_VERSION",
"READ_SCOPE",
"RUN_SCOPE",
"TRIGGER_DISPATCH_SCOPE",
"TRIGGER_READ_SCOPE",
"TRIGGER_WRITE_SCOPE",
"WRITE_SCOPE",
"get_manifest",
"manifest",
]