Add governed reusable pipelines and triggers

This commit is contained in:
2026-07-28 15:04:23 +02:00
parent 6305ef9cef
commit 08ddc8d3e8
17 changed files with 4697 additions and 46 deletions

View File

@@ -63,6 +63,24 @@ pipeline revision, and can atomically publish a complete result through
truncated; larger asynchronous and artifact-backed execution belongs to a
subsequent runner slice.
## Governed Definitions And Automation
Pipelines can be complete flows or reusable templates at system, tenant,
group, or user scope. Templates never run directly. A permitted consumer
derives a new definition that pins the source revision and content hash and
records the effective Policy decision and ancestor limits. Inherited
definitions remain read-only; lower scopes may narrow, but not broaden,
execution, reuse, inheritance, or automation permissions.
Complete active flows support explicit user/API starts, administrative
backfills, one-time schedules, interval schedules, and exact-match platform
events. Trigger deliveries are durable and idempotent. They pin the pipeline
revision and a least-privilege scope grant, then ask Access to rebuild the
owner's current principal before every run. Revoked memberships or reduced
permissions block the delivery before source access or output publication.
Confidential and restricted events are not accepted through the direct
ingress; those require Core's transactional event bridge.
## Development
```bash

View File

@@ -4,7 +4,17 @@ import uuid
from datetime import datetime
from typing import Any
from sqlalchemy import DateTime, ForeignKey, Index, Integer, JSON, String, Text, UniqueConstraint
from sqlalchemy import (
Boolean,
DateTime,
ForeignKey,
Index,
Integer,
JSON,
String,
Text,
UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from govoplan_core.db.base import Base, TimestampMixin
@@ -22,7 +32,66 @@ class DataflowPipeline(Base, TimestampMixin):
)
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)
tenant_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
scope_type: Mapped[str] = mapped_column(
String(20),
default="tenant",
nullable=False,
index=True,
)
scope_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
definition_kind: Mapped[str] = mapped_column(
String(20),
default="flow",
nullable=False,
index=True,
)
inherit_to_lower_scopes: Mapped[bool] = mapped_column(
Boolean,
default=False,
nullable=False,
)
allow_run: Mapped[bool] = mapped_column(
Boolean,
default=True,
nullable=False,
)
allow_reuse: Mapped[bool] = mapped_column(
Boolean,
default=False,
nullable=False,
)
allow_automation: Mapped[bool] = mapped_column(
Boolean,
default=False,
nullable=False,
)
derived_from_pipeline_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
derived_from_revision: Mapped[int | None] = mapped_column(
Integer,
nullable=True,
)
derived_from_hash: Mapped[str | None] = mapped_column(
String(64),
nullable=True,
)
derivation_provenance: Mapped[dict[str, Any]] = mapped_column(
JSON,
default=dict,
nullable=False,
)
name: Mapped[str] = mapped_column(String(300), nullable=False)
description: Mapped[str | None] = mapped_column(Text)
status: Mapped[str] = mapped_column(String(32), default="draft", nullable=False, index=True)
@@ -42,6 +111,11 @@ class DataflowPipeline(Base, TimestampMixin):
cascade="all, delete-orphan",
order_by="DataflowRun.created_at",
)
triggers: Mapped[list["DataflowTrigger"]] = relationship(
back_populates="pipeline",
cascade="all, delete-orphan",
order_by="DataflowTrigger.created_at",
)
class DataflowPipelineRevision(Base, TimestampMixin):
@@ -53,7 +127,11 @@ class DataflowPipelineRevision(Base, TimestampMixin):
)
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)
tenant_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
pipeline_id: Mapped[str] = mapped_column(
ForeignKey("dataflow_pipelines.id", ondelete="CASCADE"),
nullable=False,
@@ -111,6 +189,32 @@ class DataflowRun(Base, TimestampMixin):
default=dict,
nullable=False,
)
invocation_kind: Mapped[str] = mapped_column(
String(30),
default="manual",
nullable=False,
index=True,
)
trigger_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
trigger_delivery_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
correlation_id: Mapped[str | None] = mapped_column(
String(128),
nullable=True,
index=True,
)
causation_id: Mapped[str | None] = mapped_column(
String(128),
nullable=True,
index=True,
)
source_fingerprints: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
result_schema: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
diagnostics: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
@@ -136,9 +240,231 @@ class DataflowRun(Base, TimestampMixin):
pipeline: Mapped[DataflowPipeline] = relationship(back_populates="runs")
class DataflowTrigger(Base, TimestampMixin):
__tablename__ = "dataflow_triggers"
__table_args__ = (
Index(
"ix_dataflow_triggers_due",
"status",
"next_fire_at",
),
Index(
"ix_dataflow_triggers_tenant_pipeline",
"tenant_id",
"pipeline_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,
)
pipeline_id: Mapped[str] = mapped_column(
ForeignKey("dataflow_pipelines.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
pipeline_revision_id: Mapped[str] = mapped_column(
ForeignKey("dataflow_pipeline_revisions.id", ondelete="RESTRICT"),
nullable=False,
index=True,
)
name: Mapped[str] = mapped_column(String(300), nullable=False)
kind: Mapped[str] = mapped_column(
String(30),
nullable=False,
index=True,
)
status: Mapped[str] = mapped_column(
String(30),
default="disabled",
nullable=False,
index=True,
)
revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
config_: Mapped[dict[str, Any]] = mapped_column(
"config",
JSON,
default=dict,
nullable=False,
)
publication_: Mapped[dict[str, Any] | None] = mapped_column(
"publication",
JSON,
nullable=True,
)
row_limit: Mapped[int] = mapped_column(
Integer,
default=500,
nullable=False,
)
catch_up_policy: Mapped[str] = mapped_column(
String(20),
default="coalesce",
nullable=False,
)
max_concurrent_runs: Mapped[int] = mapped_column(
Integer,
default=1,
nullable=False,
)
next_fire_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
index=True,
)
last_fire_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
last_status: Mapped[str | None] = mapped_column(String(30), nullable=True)
last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
authorization_account_id: Mapped[str] = mapped_column(
String(36),
nullable=False,
)
authorization_membership_id: Mapped[str] = mapped_column(
String(36),
nullable=False,
)
authorization_ref: Mapped[str] = mapped_column(
String(255),
nullable=False,
unique=True,
)
grant_scopes: Mapped[list[str]] = mapped_column(
JSON,
default=list,
nullable=False,
)
created_by: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
index=True,
)
updated_by: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
index=True,
)
pipeline: Mapped[DataflowPipeline] = relationship(
back_populates="triggers"
)
deliveries: Mapped[list["DataflowTriggerDelivery"]] = relationship(
back_populates="trigger",
cascade="all, delete-orphan",
order_by="DataflowTriggerDelivery.created_at",
)
class DataflowTriggerDelivery(Base, TimestampMixin):
__tablename__ = "dataflow_trigger_deliveries"
__table_args__ = (
UniqueConstraint(
"trigger_id",
"source_key",
name="uq_dataflow_trigger_delivery_source",
),
Index(
"ix_dataflow_trigger_deliveries_queue",
"status",
"created_at",
),
Index(
"ix_dataflow_trigger_deliveries_tenant_trigger",
"tenant_id",
"trigger_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,
)
trigger_id: Mapped[str] = mapped_column(
ForeignKey("dataflow_triggers.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
pipeline_id: Mapped[str] = mapped_column(
ForeignKey("dataflow_pipelines.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
pipeline_revision_id: Mapped[str] = mapped_column(
ForeignKey("dataflow_pipeline_revisions.id", ondelete="RESTRICT"),
nullable=False,
index=True,
)
source_key: Mapped[str] = mapped_column(String(500), nullable=False)
invocation_kind: Mapped[str] = mapped_column(
String(30),
nullable=False,
index=True,
)
status: Mapped[str] = mapped_column(
String(30),
default="queued",
nullable=False,
index=True,
)
scheduled_for: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
event_: Mapped[dict[str, Any] | None] = mapped_column(
"event",
JSON,
nullable=True,
)
run_id: Mapped[str | None] = mapped_column(
ForeignKey("dataflow_runs.id", ondelete="SET NULL"),
nullable=True,
index=True,
)
attempts: Mapped[int] = mapped_column(
Integer,
default=0,
nullable=False,
)
authorization_provenance: Mapped[dict[str, Any]] = mapped_column(
JSON,
default=dict,
nullable=False,
)
error: Mapped[str | None] = mapped_column(Text, nullable=True)
claimed_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
finished_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
trigger: Mapped[DataflowTrigger] = relationship(
back_populates="deliveries"
)
__all__ = [
"DataflowPipeline",
"DataflowPipelineRevision",
"DataflowRun",
"DataflowTrigger",
"DataflowTriggerDelivery",
"new_uuid",
]

View File

@@ -0,0 +1,297 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import Literal
from govoplan_core.auth import ApiPrincipal, has_scope
from govoplan_core.core.policy import (
DefinitionGovernanceAction,
DefinitionGovernanceRequest,
DefinitionScopeRef,
PolicyDecision,
PolicySourceStep,
definition_governance_policy,
)
from govoplan_dataflow.backend.db.models import DataflowPipeline
DefinitionAction = Literal[
"view",
"edit",
"run",
"reuse",
"derive",
"automate",
]
GOVERNANCE_ACTIONS: tuple[DefinitionAction, ...] = (
"view",
"edit",
"run",
"reuse",
"derive",
"automate",
)
def normalize_definition_scope(
principal: ApiPrincipal,
*,
scope_type: str,
scope_id: str | None,
administrative: bool,
) -> tuple[str | None, str, str | None]:
clean_type = scope_type.strip().casefold()
clean_id = str(scope_id or "").strip() or None
if clean_type == "system":
if not has_scope(principal, "system:governance:write"):
raise PermissionError(
"System definitions require system governance permission."
)
if clean_id is not None:
raise ValueError("System definitions do not carry a scope ID.")
return None, "system", None
if clean_type == "tenant":
if clean_id not in {None, principal.tenant_id}:
raise PermissionError(
"Definitions can only be created for the active tenant."
)
return principal.tenant_id, "tenant", principal.tenant_id
if clean_type == "group":
if not clean_id:
raise ValueError("Group definitions require a group ID.")
if clean_id not in principal.group_ids and not administrative:
raise PermissionError(
"Definitions can only be created for one of the actor's groups."
)
return principal.tenant_id, "group", clean_id
if clean_type == "user":
if not clean_id:
clean_id = principal.membership_id or principal.account_id
if (
clean_id not in {principal.membership_id, principal.account_id}
and not administrative
):
raise PermissionError(
"Definitions can only be created for the current user."
)
return principal.tenant_id, "user", clean_id
raise ValueError(
"Definition scope must be system, tenant, group, or user."
)
def definition_decision(
pipeline: DataflowPipeline,
*,
principal: ApiPrincipal,
registry: object | None,
action: DefinitionGovernanceAction,
) -> PolicyDecision:
source_scope = DefinitionScopeRef(
scope_type=pipeline.scope_type, # type: ignore[arg-type]
scope_id=pipeline.scope_id,
)
request = DefinitionGovernanceRequest(
module_id="dataflow",
definition_ref=f"pipeline:{pipeline.id}",
tenant_id=principal.tenant_id,
definition_scope=source_scope,
target_scope=_target_scope(pipeline, principal),
definition_kind=pipeline.definition_kind, # type: ignore[arg-type]
action=action,
actor=principal.to_platform_principal(),
status=pipeline.status,
inherit_to_lower_scopes=pipeline.inherit_to_lower_scopes,
allow_run=pipeline.allow_run,
allow_reuse=pipeline.allow_reuse,
allow_automation=pipeline.allow_automation,
context=_ancestor_context(pipeline),
)
provider = definition_governance_policy(registry)
if provider is not None:
return provider.resolve_definition_action(request=request)
return _tenant_local_fallback(request)
def definition_governance_payload(
pipeline: DataflowPipeline,
*,
principal: ApiPrincipal,
registry: object | None,
) -> dict[str, object]:
actions = {
action: definition_decision(
pipeline,
principal=principal,
registry=registry,
action=action,
).to_dict()
for action in GOVERNANCE_ACTIONS
}
return {
"scope_type": pipeline.scope_type,
"scope_id": pipeline.scope_id,
"definition_kind": pipeline.definition_kind,
"inherit_to_lower_scopes": pipeline.inherit_to_lower_scopes,
"allow_run": pipeline.allow_run,
"allow_reuse": pipeline.allow_reuse,
"allow_automation": pipeline.allow_automation,
"derived_from_pipeline_id": pipeline.derived_from_pipeline_id,
"derived_from_revision": pipeline.derived_from_revision,
"derived_from_hash": pipeline.derived_from_hash,
"derivation_provenance": dict(pipeline.derivation_provenance),
"actions": actions,
}
def require_definition_action(
pipeline: DataflowPipeline,
*,
principal: ApiPrincipal,
registry: object | None,
action: DefinitionGovernanceAction,
) -> PolicyDecision:
decision = definition_decision(
pipeline,
principal=principal,
registry=registry,
action=action,
)
if not decision.allowed:
raise PermissionError(
decision.reason or f"Definition action is not allowed: {action}"
)
return decision
def _target_scope(
pipeline: DataflowPipeline,
principal: ApiPrincipal,
) -> DefinitionScopeRef:
if (
pipeline.scope_type == "group"
and pipeline.scope_id in principal.group_ids
):
return DefinitionScopeRef("group", pipeline.scope_id)
if pipeline.scope_type == "user" and pipeline.scope_id in {
principal.membership_id,
principal.account_id,
}:
return DefinitionScopeRef("user", pipeline.scope_id)
return DefinitionScopeRef("tenant", principal.tenant_id)
def _ancestor_context(pipeline: DataflowPipeline) -> dict[str, object]:
provenance = pipeline.derivation_provenance
limits = provenance.get("source_effective_limits")
source = provenance.get("source_scope")
context: dict[str, object] = {}
if isinstance(limits, Mapping):
context["ancestor_limits"] = dict(limits)
if isinstance(source, Mapping):
context["ancestor_source"] = dict(source)
return context
def _tenant_local_fallback(
request: DefinitionGovernanceRequest,
) -> PolicyDecision:
ancestor = request.context.get("ancestor_limits")
ancestor_limits = ancestor if isinstance(ancestor, Mapping) else {}
effective_limits = {
"inherit_to_lower_scopes": (
request.inherit_to_lower_scopes
and _fallback_ancestor_flag(
ancestor_limits,
"inherit_to_lower_scopes",
)
),
"allow_run": request.allow_run
and _fallback_ancestor_flag(ancestor_limits, "allow_run"),
"allow_reuse": request.allow_reuse
and _fallback_ancestor_flag(ancestor_limits, "allow_reuse"),
"allow_automation": request.allow_automation
and _fallback_ancestor_flag(
ancestor_limits,
"allow_automation",
),
}
local = (
request.definition_scope.scope_type == "tenant"
and request.definition_scope.scope_id == request.tenant_id
and request.actor.tenant_id == request.tenant_id
)
allowed = False
reason: str | None = None
if not local:
reason = (
"Inherited definitions require the Policy module; only local "
"tenant definitions are available."
)
elif request.action in {"view", "edit"}:
allowed = True
elif request.action == "run":
allowed = (
request.definition_kind == "flow"
and request.status == "active"
and effective_limits["allow_run"]
)
reason = (
None
if allowed
else "Only active local flows with execution enabled can run."
)
else:
reason = (
"Definition reuse and automation require the Policy module."
)
return PolicyDecision(
allowed=allowed,
reason=reason,
source_path=(
PolicySourceStep(
scope_type=request.definition_scope.scope_type,
scope_id=request.definition_scope.scope_id,
label="Tenant-local conservative fallback",
applied_fields=(
"definition_kind",
"status",
"allow_run",
),
policy={
"policy_module_available": False,
"definition_kind": request.definition_kind,
"status": request.status,
"allow_run": request.allow_run,
},
),
),
requirements=(
()
if allowed
else (f"dataflow.definition.{request.action}",)
),
details={
"fallback": "tenant_local",
"action": request.action,
"effective_limits": effective_limits,
},
)
def _fallback_ancestor_flag(
limits: Mapping[str, object],
key: str,
) -> bool:
value = limits.get(key)
return True if value is None else value is True
__all__ = [
"GOVERNANCE_ACTIONS",
"definition_decision",
"definition_governance_payload",
"normalize_definition_scope",
"require_definition_action",
]

View File

@@ -6,7 +6,13 @@ from govoplan_core.core.module_guards import (
drop_table_retirement_provider,
persistent_table_uninstall_guard,
)
from govoplan_core.core.dataflows import CAPABILITY_DATAFLOW_RUN_LIFECYCLE
from govoplan_core.core.access import (
CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
)
from govoplan_core.core.dataflows import (
CAPABILITY_DATAFLOW_RUN_LIFECYCLE,
CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
)
from govoplan_core.core.modules import (
DocumentationTopic,
FrontendModule,
@@ -24,6 +30,9 @@ from govoplan_core.core.datasources import (
CAPABILITY_DATASOURCE_LIFECYCLE,
CAPABILITY_DATASOURCE_PUBLICATION,
)
from govoplan_core.core.policy import (
CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
)
from govoplan_core.db.base import Base
from govoplan_dataflow.backend.db import models as dataflow_models
@@ -36,6 +45,9 @@ 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:
@@ -73,6 +85,21 @@ PERMISSIONS = (
"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 = (
@@ -80,19 +107,25 @@ ROLE_TEMPLATES = (
slug="dataflow_designer",
name="Dataflow designer",
description="Design, validate, and preview governed data pipelines.",
permissions=(READ_SCOPE, WRITE_SCOPE, RUN_SCOPE),
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),
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,),
permissions=(READ_SCOPE, TRIGGER_READ_SCOPE),
),
)
@@ -151,6 +184,14 @@ def _run_provider(context: ModuleContext):
return SqlDataflowRunLifecycleProvider(registry=context.registry)
def _trigger_provider(context: ModuleContext):
from govoplan_dataflow.backend.triggers import (
SqlDataflowTriggerDispatcher,
)
return SqlDataflowTriggerDispatcher(registry=context.registry)
def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
return {
"dataflow_pipelines": (
@@ -166,6 +207,19 @@ def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
.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()
),
}
@@ -189,12 +243,18 @@ manifest = ModuleManifest(
CAPABILITY_DATASOURCE_CATALOGUE,
CAPABILITY_DATASOURCE_LIFECYCLE,
CAPABILITY_DATASOURCE_PUBLICATION,
CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
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.dataset_output", version=MODULE_VERSION),
ModuleInterfaceProvider(
name="dataflow.trigger_dispatcher",
version=MODULE_VERSION,
),
),
requires_interfaces=(
ModuleInterfaceRequirement(
@@ -203,6 +263,18 @@ manifest = ModuleManifest(
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",
@@ -243,6 +315,7 @@ manifest = ModuleManifest(
route_factory=_dataflow_router,
capability_factories={
CAPABILITY_DATAFLOW_RUN_LIFECYCLE: _run_provider,
CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER: _trigger_provider,
},
tenant_summary_providers=(_tenant_summary,),
migration_spec=MigrationSpec(
@@ -251,6 +324,8 @@ manifest = ModuleManifest(
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.DataflowPipelineRevision,
dataflow_models.DataflowPipeline,
@@ -266,6 +341,8 @@ manifest = ModuleManifest(
dataflow_models.DataflowPipeline,
dataflow_models.DataflowPipelineRevision,
dataflow_models.DataflowRun,
dataflow_models.DataflowTrigger,
dataflow_models.DataflowTriggerDelivery,
label="Dataflow",
),
),
@@ -284,6 +361,9 @@ __all__ = [
"MODULE_VERSION",
"READ_SCOPE",
"RUN_SCOPE",
"TRIGGER_DISPATCH_SCOPE",
"TRIGGER_READ_SCOPE",
"TRIGGER_WRITE_SCOPE",
"WRITE_SCOPE",
"get_manifest",
"manifest",

View File

@@ -0,0 +1,433 @@
"""v0.1.14 governed definitions and Dataflow triggers
Revision ID: e2f4a8c9d1b7
Revises: b8e3c6a1f4d9
Create Date: 2026-07-28 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "e2f4a8c9d1b7"
down_revision = "b8e3c6a1f4d9"
branch_labels = None
depends_on = None
def upgrade() -> None:
with op.batch_alter_table("dataflow_pipelines") as batch_op:
batch_op.alter_column(
"tenant_id",
existing_type=sa.String(length=36),
nullable=True,
)
batch_op.add_column(
sa.Column(
"scope_type",
sa.String(length=20),
nullable=False,
server_default="tenant",
)
)
batch_op.add_column(
sa.Column("scope_id", sa.String(length=36), nullable=True)
)
batch_op.add_column(
sa.Column(
"definition_kind",
sa.String(length=20),
nullable=False,
server_default="flow",
)
)
batch_op.add_column(
sa.Column(
"inherit_to_lower_scopes",
sa.Boolean(),
nullable=False,
server_default=sa.false(),
)
)
batch_op.add_column(
sa.Column(
"allow_run",
sa.Boolean(),
nullable=False,
server_default=sa.true(),
)
)
batch_op.add_column(
sa.Column(
"allow_reuse",
sa.Boolean(),
nullable=False,
server_default=sa.false(),
)
)
batch_op.add_column(
sa.Column(
"allow_automation",
sa.Boolean(),
nullable=False,
server_default=sa.false(),
)
)
batch_op.add_column(
sa.Column(
"derived_from_pipeline_id",
sa.String(length=36),
nullable=True,
)
)
batch_op.add_column(
sa.Column(
"derived_from_revision",
sa.Integer(),
nullable=True,
)
)
batch_op.add_column(
sa.Column(
"derived_from_hash",
sa.String(length=64),
nullable=True,
)
)
batch_op.add_column(
sa.Column(
"derivation_provenance",
sa.JSON(),
nullable=False,
server_default=sa.text("'{}'"),
)
)
with op.batch_alter_table("dataflow_pipeline_revisions") as batch_op:
batch_op.alter_column(
"tenant_id",
existing_type=sa.String(length=36),
nullable=True,
)
op.execute(
sa.text(
"UPDATE dataflow_pipelines "
"SET scope_id = tenant_id "
"WHERE scope_type = 'tenant' AND scope_id IS NULL"
)
)
for column in (
"scope_type",
"scope_id",
"definition_kind",
"derived_from_pipeline_id",
):
op.create_index(
op.f(f"ix_dataflow_pipelines_{column}"),
"dataflow_pipelines",
[column],
unique=False,
)
with op.batch_alter_table("dataflow_runs") as batch_op:
batch_op.add_column(
sa.Column(
"invocation_kind",
sa.String(length=30),
nullable=False,
server_default="manual",
)
)
batch_op.add_column(
sa.Column("trigger_id", sa.String(length=36), nullable=True)
)
batch_op.add_column(
sa.Column(
"trigger_delivery_id",
sa.String(length=36),
nullable=True,
)
)
batch_op.add_column(
sa.Column(
"correlation_id",
sa.String(length=128),
nullable=True,
)
)
batch_op.add_column(
sa.Column(
"causation_id",
sa.String(length=128),
nullable=True,
)
)
for column in (
"invocation_kind",
"trigger_id",
"trigger_delivery_id",
"correlation_id",
"causation_id",
):
op.create_index(
op.f(f"ix_dataflow_runs_{column}"),
"dataflow_runs",
[column],
unique=False,
)
op.create_table(
"dataflow_triggers",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("pipeline_id", sa.String(length=36), nullable=False),
sa.Column(
"pipeline_revision_id",
sa.String(length=36),
nullable=False,
),
sa.Column("name", sa.String(length=300), nullable=False),
sa.Column("kind", sa.String(length=30), nullable=False),
sa.Column("status", sa.String(length=30), nullable=False),
sa.Column("revision", sa.Integer(), nullable=False),
sa.Column("config", sa.JSON(), nullable=False),
sa.Column("publication", sa.JSON(), nullable=True),
sa.Column("row_limit", sa.Integer(), nullable=False),
sa.Column("catch_up_policy", sa.String(length=20), nullable=False),
sa.Column("max_concurrent_runs", sa.Integer(), nullable=False),
sa.Column("next_fire_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("last_fire_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("last_status", sa.String(length=30), nullable=True),
sa.Column("last_error", sa.Text(), nullable=True),
sa.Column(
"authorization_account_id",
sa.String(length=36),
nullable=False,
),
sa.Column(
"authorization_membership_id",
sa.String(length=36),
nullable=False,
),
sa.Column(
"authorization_ref",
sa.String(length=255),
nullable=False,
),
sa.Column("grant_scopes", sa.JSON(), nullable=False),
sa.Column("created_by", sa.String(length=255), nullable=True),
sa.Column("updated_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(
["pipeline_id"],
["dataflow_pipelines.id"],
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["pipeline_revision_id"],
["dataflow_pipeline_revisions.id"],
ondelete="RESTRICT",
),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint(
"authorization_ref",
name="uq_dataflow_triggers_authorization_ref",
),
)
for column in (
"tenant_id",
"pipeline_id",
"pipeline_revision_id",
"kind",
"status",
"next_fire_at",
"created_by",
"updated_by",
):
op.create_index(
op.f(f"ix_dataflow_triggers_{column}"),
"dataflow_triggers",
[column],
unique=False,
)
op.create_index(
"ix_dataflow_triggers_due",
"dataflow_triggers",
["status", "next_fire_at"],
unique=False,
)
op.create_index(
"ix_dataflow_triggers_tenant_pipeline",
"dataflow_triggers",
["tenant_id", "pipeline_id"],
unique=False,
)
op.create_table(
"dataflow_trigger_deliveries",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("trigger_id", sa.String(length=36), nullable=False),
sa.Column("pipeline_id", sa.String(length=36), nullable=False),
sa.Column(
"pipeline_revision_id",
sa.String(length=36),
nullable=False,
),
sa.Column("source_key", sa.String(length=500), nullable=False),
sa.Column("invocation_kind", sa.String(length=30), nullable=False),
sa.Column("status", sa.String(length=30), nullable=False),
sa.Column("scheduled_for", sa.DateTime(timezone=True), nullable=True),
sa.Column("event", sa.JSON(), nullable=True),
sa.Column("run_id", sa.String(length=36), nullable=True),
sa.Column("attempts", sa.Integer(), nullable=False),
sa.Column(
"authorization_provenance",
sa.JSON(),
nullable=False,
),
sa.Column("error", sa.Text(), nullable=True),
sa.Column("claimed_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(
["pipeline_id"],
["dataflow_pipelines.id"],
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["pipeline_revision_id"],
["dataflow_pipeline_revisions.id"],
ondelete="RESTRICT",
),
sa.ForeignKeyConstraint(
["run_id"],
["dataflow_runs.id"],
ondelete="SET NULL",
),
sa.ForeignKeyConstraint(
["trigger_id"],
["dataflow_triggers.id"],
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint(
"trigger_id",
"source_key",
name="uq_dataflow_trigger_delivery_source",
),
)
for column in (
"tenant_id",
"trigger_id",
"pipeline_id",
"pipeline_revision_id",
"invocation_kind",
"status",
"run_id",
):
op.create_index(
op.f(f"ix_dataflow_trigger_deliveries_{column}"),
"dataflow_trigger_deliveries",
[column],
unique=False,
)
op.create_index(
"ix_dataflow_trigger_deliveries_queue",
"dataflow_trigger_deliveries",
["status", "created_at"],
unique=False,
)
op.create_index(
"ix_dataflow_trigger_deliveries_tenant_trigger",
"dataflow_trigger_deliveries",
["tenant_id", "trigger_id"],
unique=False,
)
def downgrade() -> None:
op.drop_index(
"ix_dataflow_trigger_deliveries_tenant_trigger",
table_name="dataflow_trigger_deliveries",
)
op.drop_index(
"ix_dataflow_trigger_deliveries_queue",
table_name="dataflow_trigger_deliveries",
)
op.drop_table("dataflow_trigger_deliveries")
op.drop_index(
"ix_dataflow_triggers_tenant_pipeline",
table_name="dataflow_triggers",
)
op.drop_index(
"ix_dataflow_triggers_due",
table_name="dataflow_triggers",
)
op.drop_table("dataflow_triggers")
for column in (
"causation_id",
"correlation_id",
"trigger_delivery_id",
"trigger_id",
"invocation_kind",
):
op.drop_index(
op.f(f"ix_dataflow_runs_{column}"),
table_name="dataflow_runs",
)
with op.batch_alter_table("dataflow_runs") as batch_op:
batch_op.drop_column("causation_id")
batch_op.drop_column("correlation_id")
batch_op.drop_column("trigger_delivery_id")
batch_op.drop_column("trigger_id")
batch_op.drop_column("invocation_kind")
for column in (
"derived_from_pipeline_id",
"definition_kind",
"scope_id",
"scope_type",
):
op.drop_index(
op.f(f"ix_dataflow_pipelines_{column}"),
table_name="dataflow_pipelines",
)
op.execute(
sa.text(
"UPDATE dataflow_pipeline_revisions "
"SET tenant_id = COALESCE(tenant_id, 'system')"
)
)
op.execute(
sa.text(
"UPDATE dataflow_pipelines "
"SET tenant_id = COALESCE(tenant_id, 'system')"
)
)
with op.batch_alter_table("dataflow_pipelines") as batch_op:
batch_op.drop_column("derivation_provenance")
batch_op.drop_column("derived_from_hash")
batch_op.drop_column("derived_from_revision")
batch_op.drop_column("derived_from_pipeline_id")
batch_op.drop_column("allow_automation")
batch_op.drop_column("allow_reuse")
batch_op.drop_column("allow_run")
batch_op.drop_column("inherit_to_lower_scopes")
batch_op.drop_column("definition_kind")
batch_op.drop_column("scope_id")
batch_op.drop_column("scope_type")
batch_op.alter_column(
"tenant_id",
existing_type=sa.String(length=36),
nullable=False,
)
with op.batch_alter_table("dataflow_pipeline_revisions") as batch_op:
batch_op.alter_column(
"tenant_id",
existing_type=sa.String(length=36),
nullable=False,
)

View File

@@ -5,6 +5,7 @@ from sqlalchemy.orm import Session
from govoplan_core.audit.logging import audit_event
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
from govoplan_core.core.automation import AutomationInvocation
from govoplan_core.core.dataflows import (
DataflowPublicationTarget,
DataflowRunRequest,
@@ -24,13 +25,27 @@ from govoplan_core.core.tabular_sources import (
parse_tabular_csv,
)
from govoplan_core.db.session import get_session
from govoplan_dataflow.backend.manifest import ADMIN_SCOPE, READ_SCOPE, RUN_SCOPE, WRITE_SCOPE
from govoplan_dataflow.backend.governance import (
definition_decision,
normalize_definition_scope,
require_definition_action,
)
from govoplan_dataflow.backend.manifest import (
ADMIN_SCOPE,
READ_SCOPE,
RUN_SCOPE,
TRIGGER_DISPATCH_SCOPE,
TRIGGER_READ_SCOPE,
TRIGGER_WRITE_SCOPE,
WRITE_SCOPE,
)
from govoplan_dataflow.backend.node_library import CATEGORY_LABELS, NODE_LIBRARY
from govoplan_dataflow.backend.runtime import get_registry
from govoplan_dataflow.backend.schemas import (
NodeLibraryResponse,
NodeTypeDefinitionResponse,
PipelineCreateRequest,
PipelineDeriveRequest,
PipelineDeleteResponse,
PipelineDraftRequest,
PipelineListResponse,
@@ -47,6 +62,14 @@ from govoplan_dataflow.backend.schemas import (
TabularSourceColumnResponse,
TabularSourceListResponse,
TabularSourceResponse,
DataflowEventDeliveryRequest,
DataflowTriggerCreateRequest,
DataflowTriggerDeliveryListResponse,
DataflowTriggerDeleteResponse,
DataflowTriggerDispatchResponse,
DataflowTriggerListResponse,
DataflowTriggerResponse,
DataflowTriggerUpdateRequest,
)
from govoplan_dataflow.backend.service import (
DataflowConflictError,
@@ -55,6 +78,7 @@ from govoplan_dataflow.backend.service import (
DataflowValidationError,
compile_sql_draft,
create_pipeline,
derive_pipeline,
cancel_pipeline_run,
delete_pipeline,
get_pipeline,
@@ -69,6 +93,17 @@ from govoplan_dataflow.backend.service import (
update_pipeline,
validate_draft,
)
from govoplan_dataflow.backend.triggers import (
create_trigger,
delete_trigger,
delivery_response,
dispatch_due_triggers,
ingest_event_delivery,
list_deliveries,
list_triggers,
trigger_response,
update_trigger,
)
router = APIRouter(prefix="/dataflow", tags=["dataflow"])
@@ -108,6 +143,30 @@ def _http_error(exc: DataflowError) -> HTTPException:
return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
def _governance_http_error(exc: PermissionError | ValueError) -> HTTPException:
return HTTPException(
status_code=(
status.HTTP_403_FORBIDDEN
if isinstance(exc, PermissionError)
else status.HTTP_422_UNPROCESSABLE_CONTENT
),
detail=str(exc),
)
def _pipeline_response(
session: Session,
pipeline,
principal: ApiPrincipal,
) -> PipelineResponse:
return pipeline_response(
session,
pipeline,
principal=principal,
registry=get_registry(),
)
def _source_http_error(exc: DatasourceError) -> HTTPException:
if isinstance(exc, DatasourceAccessError):
return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
@@ -317,8 +376,27 @@ def api_list_pipelines(
) -> PipelineListResponse:
_require_any_scope(principal, READ_SCOPE, ADMIN_SCOPE)
pipelines = list_pipelines(session, tenant_id=principal.tenant_id)
registry = get_registry()
visible = [
pipeline
for pipeline in pipelines
if definition_decision(
pipeline,
principal=principal,
registry=registry,
action="view",
).allowed
]
return PipelineListResponse(
pipelines=[pipeline_response(session, pipeline) for pipeline in pipelines]
pipelines=[
pipeline_response(
session,
pipeline,
principal=principal,
registry=registry,
)
for pipeline in visible
]
)
@@ -330,12 +408,23 @@ def api_create_pipeline(
) -> PipelineResponse:
_require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
try:
tenant_id, scope_type, scope_id = normalize_definition_scope(
principal,
scope_type=payload.scope_type,
scope_id=payload.scope_id,
administrative=has_scope(principal, ADMIN_SCOPE),
)
payload = payload.model_copy(
update={"scope_type": scope_type, "scope_id": scope_id}
)
pipeline = create_pipeline(
session,
tenant_id=principal.tenant_id,
tenant_id=tenant_id or principal.tenant_id,
actor_id=_actor_id(principal),
payload=payload,
)
except (PermissionError, ValueError) as exc:
raise _governance_http_error(exc) from exc
except DataflowError as exc:
raise _http_error(exc) from exc
audit_event(
@@ -348,7 +437,7 @@ def api_create_pipeline(
object_id=pipeline.id,
details={"revision": pipeline.current_revision, "status": pipeline.status},
)
response = pipeline_response(session, pipeline)
response = _pipeline_response(session, pipeline, principal)
session.commit()
return response
@@ -366,7 +455,15 @@ def api_get_pipeline(
tenant_id=principal.tenant_id,
pipeline_id=pipeline_id,
)
return pipeline_response(session, pipeline)
require_definition_action(
pipeline,
principal=principal,
registry=get_registry(),
action="view",
)
return _pipeline_response(session, pipeline, principal)
except PermissionError as exc:
raise _governance_http_error(exc) from exc
except DataflowError as exc:
raise _http_error(exc) from exc
@@ -380,6 +477,17 @@ def api_update_pipeline(
) -> PipelineResponse:
_require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
try:
existing = get_pipeline(
session,
tenant_id=principal.tenant_id,
pipeline_id=pipeline_id,
)
require_definition_action(
existing,
principal=principal,
registry=get_registry(),
action="edit",
)
pipeline = update_pipeline(
session,
tenant_id=principal.tenant_id,
@@ -387,6 +495,8 @@ def api_update_pipeline(
actor_id=_actor_id(principal),
payload=payload,
)
except PermissionError as exc:
raise _governance_http_error(exc) from exc
except DataflowError as exc:
raise _http_error(exc) from exc
audit_event(
@@ -399,7 +509,7 @@ def api_update_pipeline(
object_id=pipeline.id,
details={"revision": pipeline.current_revision, "status": pipeline.status},
)
response = pipeline_response(session, pipeline)
response = _pipeline_response(session, pipeline, principal)
session.commit()
return response
@@ -412,12 +522,25 @@ def api_delete_pipeline(
) -> PipelineDeleteResponse:
_require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
try:
existing = get_pipeline(
session,
tenant_id=principal.tenant_id,
pipeline_id=pipeline_id,
)
require_definition_action(
existing,
principal=principal,
registry=get_registry(),
action="edit",
)
pipeline = delete_pipeline(
session,
tenant_id=principal.tenant_id,
pipeline_id=pipeline_id,
actor_id=_actor_id(principal),
)
except PermissionError as exc:
raise _governance_http_error(exc) from exc
except DataflowError as exc:
raise _http_error(exc) from exc
audit_event(
@@ -434,6 +557,314 @@ def api_delete_pipeline(
return PipelineDeleteResponse(deleted=True, pipeline_id=pipeline.id)
@router.post(
"/pipelines/{pipeline_id}/derive",
response_model=PipelineResponse,
status_code=status.HTTP_201_CREATED,
)
def api_derive_pipeline(
pipeline_id: str,
payload: PipelineDeriveRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> PipelineResponse:
_require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
try:
tenant_id, scope_type, scope_id = normalize_definition_scope(
principal,
scope_type=payload.scope_type,
scope_id=payload.scope_id,
administrative=has_scope(principal, ADMIN_SCOPE),
)
payload = payload.model_copy(
update={"scope_type": scope_type, "scope_id": scope_id}
)
pipeline = derive_pipeline(
session,
tenant_id=tenant_id or principal.tenant_id,
actor_id=_actor_id(principal),
principal=principal,
registry=get_registry(),
source_pipeline_id=pipeline_id,
payload=payload,
)
except (PermissionError, ValueError) as exc:
raise _governance_http_error(exc) from exc
except DataflowError as exc:
raise _http_error(exc) from exc
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=getattr(principal.user, "id", None),
api_key_id=principal.api_key_id,
action="dataflow.pipeline.derived",
object_type="dataflow_pipeline",
object_id=pipeline.id,
details={
"source_pipeline_id": pipeline.derived_from_pipeline_id,
"source_revision": pipeline.derived_from_revision,
"source_hash": pipeline.derived_from_hash,
"scope_type": pipeline.scope_type,
"scope_id": pipeline.scope_id,
},
)
response = _pipeline_response(session, pipeline, principal)
session.commit()
return response
@router.get(
"/pipelines/{pipeline_id}/triggers",
response_model=DataflowTriggerListResponse,
)
def api_list_triggers(
pipeline_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DataflowTriggerListResponse:
_require_any_scope(principal, TRIGGER_READ_SCOPE, ADMIN_SCOPE)
try:
pipeline = get_pipeline(
session,
tenant_id=principal.tenant_id,
pipeline_id=pipeline_id,
)
require_definition_action(
pipeline,
principal=principal,
registry=get_registry(),
action="view",
)
items = list_triggers(
session,
tenant_id=principal.tenant_id,
pipeline_id=pipeline_id,
)
return DataflowTriggerListResponse(
triggers=[trigger_response(session, item) for item in items]
)
except PermissionError as exc:
raise _governance_http_error(exc) from exc
except DataflowError as exc:
raise _http_error(exc) from exc
@router.post(
"/pipelines/{pipeline_id}/triggers",
response_model=DataflowTriggerResponse,
status_code=status.HTTP_201_CREATED,
)
def api_create_trigger(
pipeline_id: str,
payload: DataflowTriggerCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DataflowTriggerResponse:
_require_any_scope(principal, TRIGGER_WRITE_SCOPE, ADMIN_SCOPE)
try:
trigger = create_trigger(
session,
tenant_id=principal.tenant_id,
pipeline_id=pipeline_id,
actor_id=_actor_id(principal),
principal=principal,
registry=get_registry(),
payload=payload,
)
except DataflowError as exc:
raise _http_error(exc) from exc
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=getattr(principal.user, "id", None),
api_key_id=principal.api_key_id,
action="dataflow.trigger.created",
object_type="dataflow_trigger",
object_id=trigger.id,
details={
"pipeline_id": trigger.pipeline_id,
"kind": trigger.kind,
"status": trigger.status,
"grant_scopes": list(trigger.grant_scopes),
},
)
response = trigger_response(session, trigger)
session.commit()
return response
@router.put(
"/triggers/{trigger_id}",
response_model=DataflowTriggerResponse,
)
def api_update_trigger(
trigger_id: str,
payload: DataflowTriggerUpdateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DataflowTriggerResponse:
_require_any_scope(principal, TRIGGER_WRITE_SCOPE, ADMIN_SCOPE)
try:
trigger = update_trigger(
session,
tenant_id=principal.tenant_id,
trigger_id=trigger_id,
actor_id=_actor_id(principal),
principal=principal,
registry=get_registry(),
payload=payload,
)
except DataflowError as exc:
raise _http_error(exc) from exc
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=getattr(principal.user, "id", None),
api_key_id=principal.api_key_id,
action="dataflow.trigger.updated",
object_type="dataflow_trigger",
object_id=trigger.id,
details={
"pipeline_id": trigger.pipeline_id,
"kind": trigger.kind,
"status": trigger.status,
"revision": trigger.revision,
"grant_scopes": list(trigger.grant_scopes),
},
)
response = trigger_response(session, trigger)
session.commit()
return response
@router.delete(
"/triggers/{trigger_id}",
response_model=DataflowTriggerDeleteResponse,
)
def api_delete_trigger(
trigger_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DataflowTriggerDeleteResponse:
_require_any_scope(principal, TRIGGER_WRITE_SCOPE, ADMIN_SCOPE)
try:
trigger = delete_trigger(
session,
tenant_id=principal.tenant_id,
trigger_id=trigger_id,
)
except DataflowError as exc:
raise _http_error(exc) from exc
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=getattr(principal.user, "id", None),
api_key_id=principal.api_key_id,
action="dataflow.trigger.deleted",
object_type="dataflow_trigger",
object_id=trigger.id,
details={"pipeline_id": trigger.pipeline_id},
)
session.commit()
return DataflowTriggerDeleteResponse(
deleted=True,
trigger_id=trigger.id,
pipeline_id=trigger.pipeline_id,
)
@router.get(
"/triggers/{trigger_id}/deliveries",
response_model=DataflowTriggerDeliveryListResponse,
)
def api_list_trigger_deliveries(
trigger_id: str,
limit: int = 50,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DataflowTriggerDeliveryListResponse:
_require_any_scope(principal, TRIGGER_READ_SCOPE, ADMIN_SCOPE)
try:
items = list_deliveries(
session,
tenant_id=principal.tenant_id,
trigger_id=trigger_id,
limit=limit,
)
except DataflowError as exc:
raise _http_error(exc) from exc
return DataflowTriggerDeliveryListResponse(
deliveries=[delivery_response(item) for item in items]
)
@router.post(
"/triggers/events",
response_model=DataflowTriggerDispatchResponse,
)
def api_ingest_trigger_event(
payload: DataflowEventDeliveryRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DataflowTriggerDispatchResponse:
_require_any_scope(principal, TRIGGER_DISPATCH_SCOPE, ADMIN_SCOPE)
try:
queued = ingest_event_delivery(
session,
tenant_id=principal.tenant_id,
event=payload,
)
except DataflowError as exc:
raise _http_error(exc) from exc
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=getattr(principal.user, "id", None),
api_key_id=principal.api_key_id,
action="dataflow.trigger.event_ingested",
object_type="platform_event",
object_id=payload.event_id,
details={
"event_type": payload.type,
"module_id": payload.module_id,
"classification": payload.classification,
"queued": queued,
},
)
session.commit()
return DataflowTriggerDispatchResponse(queued=queued)
@router.post(
"/triggers/dispatch",
response_model=DataflowTriggerDispatchResponse,
)
def api_dispatch_triggers(
limit: int = 50,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DataflowTriggerDispatchResponse:
_require_any_scope(principal, TRIGGER_DISPATCH_SCOPE, ADMIN_SCOPE)
result = dispatch_due_triggers(
session,
registry=get_registry(),
limit=limit,
tenant_id=principal.tenant_id,
)
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=getattr(principal.user, "id", None),
api_key_id=principal.api_key_id,
action="dataflow.trigger.dispatched",
object_type="dataflow_trigger_dispatch",
object_id=principal.tenant_id,
details=result.model_dump(mode="json"),
)
session.commit()
return result
@router.get(
"/pipelines/{pipeline_id}/runs",
response_model=PipelineRunListResponse,
@@ -446,6 +877,17 @@ def api_list_pipeline_runs(
) -> PipelineRunListResponse:
_require_any_scope(principal, READ_SCOPE, RUN_SCOPE, ADMIN_SCOPE)
try:
pipeline = get_pipeline(
session,
tenant_id=principal.tenant_id,
pipeline_id=pipeline_id,
)
require_definition_action(
pipeline,
principal=principal,
registry=get_registry(),
action="view",
)
runs = list_pipeline_runs(
session,
tenant_id=principal.tenant_id,
@@ -455,6 +897,8 @@ def api_list_pipeline_runs(
return PipelineRunListResponse(
runs=[pipeline_run_response(session, run) for run in runs]
)
except PermissionError as exc:
raise _governance_http_error(exc) from exc
except DataflowError as exc:
raise _http_error(exc) from exc
@@ -471,6 +915,8 @@ def api_start_pipeline_run(
principal: ApiPrincipal = Depends(get_api_principal),
) -> PipelineRunResponse:
_require_any_scope(principal, RUN_SCOPE, ADMIN_SCOPE)
if payload.invocation_kind == "backfill":
_require_any_scope(principal, ADMIN_SCOPE)
try:
pipeline = get_pipeline(
session,
@@ -503,6 +949,16 @@ def api_start_pipeline_run(
idempotency_key=payload.idempotency_key,
row_limit=payload.row_limit,
publication=publication,
invocation=AutomationInvocation(
kind=(
"backfill"
if payload.invocation_kind == "backfill"
else "api"
if principal.auth_method == "api_key"
else "manual"
),
requested_by=_actor_id(principal),
),
),
)
except DataflowError as exc:
@@ -523,6 +979,7 @@ def api_start_pipeline_run(
"pipeline_id": run.pipeline_id,
"revision_id": run.pipeline_revision_id,
"status": run.status,
"invocation_kind": run.invocation_kind,
"output_datasource_ref": run.output_datasource_ref,
"output_materialization_ref": run.output_materialization_ref,
},

View File

@@ -8,8 +8,20 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator, model_valida
PipelineStatus = Literal["draft", "active", "archived"]
DefinitionScopeType = Literal["system", "tenant", "group", "user"]
DefinitionKind = Literal["flow", "template"]
EditorMode = Literal["graph", "sql"]
DiagnosticSeverity = Literal["info", "warning", "error"]
TriggerKind = Literal["once", "interval", "event"]
TriggerStatus = Literal["enabled", "disabled", "completed", "blocked"]
TriggerDeliveryStatus = Literal[
"queued",
"running",
"succeeded",
"failed",
"blocked",
"skipped",
]
class GraphPosition(BaseModel):
@@ -68,9 +80,32 @@ class PipelineRevisionResponse(BaseModel):
created_at: datetime
class DefinitionActionDecisionResponse(BaseModel):
allowed: bool
reason: str | None = None
source_path: list[dict[str, Any]] = Field(default_factory=list)
requirements: list[str] = Field(default_factory=list)
details: dict[str, Any] = Field(default_factory=dict)
class PipelineGovernanceResponse(BaseModel):
scope_type: DefinitionScopeType
scope_id: str | None
definition_kind: DefinitionKind
inherit_to_lower_scopes: bool
allow_run: bool
allow_reuse: bool
allow_automation: bool
derived_from_pipeline_id: str | None
derived_from_revision: int | None
derived_from_hash: str | None
derivation_provenance: dict[str, Any] = Field(default_factory=dict)
actions: dict[str, DefinitionActionDecisionResponse]
class PipelineResponse(BaseModel):
id: str
tenant_id: str
tenant_id: str | None
name: str
description: str | None
status: PipelineStatus
@@ -80,6 +115,7 @@ class PipelineResponse(BaseModel):
created_at: datetime
updated_at: datetime
revision: PipelineRevisionResponse
governance: PipelineGovernanceResponse
class PipelineListResponse(BaseModel):
@@ -93,12 +129,38 @@ class PipelineCreateRequest(BaseModel):
graph: PipelineGraph
sql_text: str | None = Field(default=None, max_length=100_000)
editor_mode: EditorMode = "graph"
scope_type: DefinitionScopeType = "tenant"
scope_id: str | None = Field(default=None, max_length=36)
definition_kind: DefinitionKind = "flow"
inherit_to_lower_scopes: bool = False
allow_run: bool = True
allow_reuse: bool = False
allow_automation: bool = False
@model_validator(mode="after")
def validate_definition_kind(self) -> "PipelineCreateRequest":
if self.definition_kind == "template" and self.status == "active":
raise ValueError("Templates cannot be active.")
return self
class PipelineUpdateRequest(PipelineCreateRequest):
expected_revision: int = Field(ge=1)
class PipelineDeriveRequest(BaseModel):
name: str = Field(min_length=1, max_length=300)
description: str | None = Field(default=None, max_length=4_000)
source_revision: int | None = Field(default=None, ge=1)
scope_type: DefinitionScopeType = "tenant"
scope_id: str | None = Field(default=None, max_length=36)
definition_kind: DefinitionKind = "flow"
inherit_to_lower_scopes: bool = False
allow_run: bool = True
allow_reuse: bool = False
allow_automation: bool = False
class PipelineDraftRequest(BaseModel):
graph: PipelineGraph | None = None
sql_text: str | None = Field(default=None, max_length=100_000)
@@ -189,6 +251,7 @@ class PipelineRunCreateRequest(BaseModel):
idempotency_key: str = Field(min_length=1, max_length=255)
row_limit: int = Field(default=500, ge=1, le=500)
publication: PipelineRunPublicationRequest | None = None
invocation_kind: Literal["manual", "api", "backfill"] = "manual"
class PipelineRunResponse(BaseModel):
@@ -208,6 +271,11 @@ class PipelineRunResponse(BaseModel):
output_publication_ref: str | None
output_datasource_ref: str | None
output_materialization_ref: str | None
invocation_kind: str
trigger_ref: str | None
delivery_ref: str | None
correlation_id: str | None
causation_id: str | None
error: str | None
started_at: datetime
finished_at: datetime | None
@@ -220,6 +288,168 @@ class PipelineRunListResponse(BaseModel):
runs: list[PipelineRunResponse]
JsonScalar = str | int | float | bool | None
class DataflowTriggerSchedule(BaseModel):
run_at: datetime | None = None
anchor_at: datetime | None = None
interval_seconds: int | None = Field(default=None, ge=60, le=31_536_000)
timezone: str = Field(default="UTC", min_length=1, max_length=100)
class DataflowTriggerEvent(BaseModel):
event_type: str = Field(
min_length=1,
max_length=200,
pattern=r"^[A-Za-z0-9_.:-]+$",
)
module_id: str | None = Field(
default=None,
max_length=100,
pattern=r"^[a-z][a-z0-9_]*$",
)
filters: dict[str, JsonScalar] = Field(default_factory=dict)
@field_validator("filters")
@classmethod
def bounded_filters(
cls,
value: dict[str, JsonScalar],
) -> dict[str, JsonScalar]:
if len(value) > 20:
raise ValueError("Event triggers support at most 20 filters.")
for key in value:
if not key or len(key) > 120 or "." in key:
raise ValueError(
"Event filter keys must be direct payload fields."
)
return value
class DataflowTriggerCreateRequest(BaseModel):
name: str = Field(min_length=1, max_length=300)
kind: TriggerKind
revision: int | None = Field(default=None, ge=1)
enabled: bool = False
schedule: DataflowTriggerSchedule | None = None
event: DataflowTriggerEvent | None = None
publication: PipelineRunPublicationRequest | None = None
row_limit: int = Field(default=500, ge=1, le=500)
catch_up_policy: Literal["skip", "coalesce"] = "coalesce"
max_concurrent_runs: int = Field(default=1, ge=1, le=10)
@model_validator(mode="after")
def validate_trigger(self) -> "DataflowTriggerCreateRequest":
if self.kind == "once":
if self.schedule is None or self.schedule.run_at is None:
raise ValueError("One-time triggers require schedule.run_at.")
elif self.kind == "interval":
if (
self.schedule is None
or self.schedule.interval_seconds is None
):
raise ValueError(
"Interval triggers require schedule.interval_seconds."
)
elif self.kind == "event" and self.event is None:
raise ValueError("Event triggers require event configuration.")
return self
class DataflowTriggerUpdateRequest(DataflowTriggerCreateRequest):
expected_revision: int = Field(ge=1)
class DataflowTriggerResponse(BaseModel):
id: str
tenant_id: str
pipeline_id: str
pipeline_revision: int
name: str
kind: TriggerKind
status: TriggerStatus
revision: int
schedule: DataflowTriggerSchedule | None
event: DataflowTriggerEvent | None
publication: PipelineRunPublicationRequest | None
row_limit: int
catch_up_policy: Literal["skip", "coalesce"]
max_concurrent_runs: int
next_fire_at: datetime | None
last_fire_at: datetime | None
last_status: str | None
last_error: str | None
grant_scopes: list[str]
authorization_provenance: dict[str, Any] = Field(default_factory=dict)
created_by: str | None
updated_by: str | None
created_at: datetime
updated_at: datetime
class DataflowTriggerListResponse(BaseModel):
triggers: list[DataflowTriggerResponse]
class DataflowTriggerDeliveryResponse(BaseModel):
id: str
trigger_id: str
invocation_kind: str
status: TriggerDeliveryStatus
scheduled_for: datetime | None
event_id: str | None
run_id: str | None
attempts: int
authorization_provenance: dict[str, Any] = Field(default_factory=dict)
error: str | None
created_at: datetime
finished_at: datetime | None
class DataflowTriggerDeliveryListResponse(BaseModel):
deliveries: list[DataflowTriggerDeliveryResponse]
class DataflowEventDeliveryRequest(BaseModel):
event_id: str = Field(min_length=1, max_length=128)
type: str = Field(
min_length=1,
max_length=200,
pattern=r"^[A-Za-z0-9_.:-]+$",
)
module_id: str = Field(
min_length=1,
max_length=100,
pattern=r"^[a-z][a-z0-9_]*$",
)
payload: dict[str, Any] = Field(default_factory=dict)
occurred_at: datetime
correlation_id: str | None = Field(default=None, max_length=128)
causation_id: str | None = Field(default=None, max_length=128)
classification: Literal[
"public",
"internal",
"confidential",
"restricted",
] = "internal"
class DataflowTriggerDispatchResponse(BaseModel):
queued: int = 0
processed: int = 0
succeeded: int = 0
failed: int = 0
blocked: int = 0
skipped: int = 0
class DataflowTriggerDeleteResponse(BaseModel):
deleted: bool
trigger_id: str
pipeline_id: str
class PipelineDeleteResponse(BaseModel):
deleted: bool
pipeline_id: str

View File

@@ -1,13 +1,15 @@
from __future__ import annotations
from collections.abc import Mapping
import hashlib
import json
from dataclasses import dataclass
from sqlalchemy import select
from sqlalchemy import or_, select
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.automation import AutomationInvocation
from govoplan_core.core.dataflows import (
DataflowRunConflictError,
DataflowRunDescriptor,
@@ -34,11 +36,16 @@ from govoplan_dataflow.backend.executor import (
ResolvedSource,
execute_preview,
)
from govoplan_dataflow.backend.governance import (
definition_governance_payload,
require_definition_action,
)
from govoplan_dataflow.backend.graph import canonical_graph_payload, definition_hash, validate_graph
from govoplan_dataflow.backend.schemas import (
DataflowDiagnostic,
GraphNode,
PipelineCreateRequest,
PipelineDeriveRequest,
PipelineDraftRequest,
PipelineGraph,
PipelinePreviewRequest,
@@ -87,7 +94,10 @@ def list_pipelines(session: Session, *, tenant_id: str) -> list[DataflowPipeline
session.scalars(
select(DataflowPipeline)
.where(
DataflowPipeline.tenant_id == tenant_id,
or_(
DataflowPipeline.tenant_id == tenant_id,
DataflowPipeline.tenant_id.is_(None),
),
DataflowPipeline.deleted_at.is_(None),
)
.order_by(DataflowPipeline.updated_at.desc(), DataflowPipeline.name)
@@ -104,7 +114,10 @@ def get_pipeline(
pipeline = session.scalar(
select(DataflowPipeline).where(
DataflowPipeline.id == pipeline_id,
DataflowPipeline.tenant_id == tenant_id,
or_(
DataflowPipeline.tenant_id == tenant_id,
DataflowPipeline.tenant_id.is_(None),
),
DataflowPipeline.deleted_at.is_(None),
)
)
@@ -145,8 +158,23 @@ def create_pipeline(
editor_mode=payload.editor_mode,
)
content_hash = definition_hash(definition.graph, definition.sql_text)
stored_tenant_id = None if payload.scope_type == "system" else tenant_id
scope_id = (
None
if payload.scope_type == "system"
else tenant_id
if payload.scope_type == "tenant"
else payload.scope_id
)
pipeline = DataflowPipeline(
tenant_id=tenant_id,
tenant_id=stored_tenant_id,
scope_type=payload.scope_type,
scope_id=scope_id,
definition_kind=payload.definition_kind,
inherit_to_lower_scopes=payload.inherit_to_lower_scopes,
allow_run=payload.allow_run,
allow_reuse=payload.allow_reuse,
allow_automation=payload.allow_automation,
name=payload.name.strip(),
description=_clean_optional(payload.description),
status=payload.status,
@@ -156,7 +184,7 @@ def create_pipeline(
metadata_={},
)
revision = DataflowPipelineRevision(
tenant_id=tenant_id,
tenant_id=stored_tenant_id,
revision=1,
schema_version=definition.graph.schema_version,
graph=canonical_graph_payload(definition.graph),
@@ -185,6 +213,21 @@ def update_pipeline(
f"Pipeline changed on the server; expected revision {payload.expected_revision}, "
f"current revision is {pipeline.current_revision}"
)
if (
payload.scope_type != pipeline.scope_type
or payload.scope_id != pipeline.scope_id
and not (
pipeline.scope_type == "tenant"
and payload.scope_id in {None, pipeline.scope_id}
)
):
raise DataflowConflictError(
"Definition scope is immutable; derive a scoped copy instead."
)
if payload.definition_kind != pipeline.definition_kind:
raise DataflowConflictError(
"Definition kind is immutable; derive a flow or template instead."
)
definition = normalize_definition(
graph=payload.graph,
sql_text=payload.sql_text,
@@ -195,12 +238,26 @@ def update_pipeline(
pipeline.name = payload.name.strip()
pipeline.description = _clean_optional(payload.description)
pipeline.status = payload.status
ancestor_limits = _ancestor_governance_limits(
pipeline.derivation_provenance
)
pipeline.inherit_to_lower_scopes = (
payload.inherit_to_lower_scopes
and ancestor_limits["inherit_to_lower_scopes"]
)
pipeline.allow_run = payload.allow_run and ancestor_limits["allow_run"]
pipeline.allow_reuse = (
payload.allow_reuse and ancestor_limits["allow_reuse"]
)
pipeline.allow_automation = (
payload.allow_automation and ancestor_limits["allow_automation"]
)
pipeline.updated_by = actor_id
if current.content_hash != content_hash or current.editor_mode != payload.editor_mode:
pipeline.current_revision += 1
pipeline.revisions.append(
DataflowPipelineRevision(
tenant_id=tenant_id,
tenant_id=pipeline.tenant_id,
revision=pipeline.current_revision,
schema_version=definition.graph.schema_version,
graph=canonical_graph_payload(definition.graph),
@@ -214,6 +271,110 @@ def update_pipeline(
return pipeline
def derive_pipeline(
session: Session,
*,
tenant_id: str,
actor_id: str | None,
principal: ApiPrincipal,
registry: object | None,
source_pipeline_id: str,
payload: PipelineDeriveRequest,
) -> DataflowPipeline:
source = get_pipeline(
session,
tenant_id=tenant_id,
pipeline_id=source_pipeline_id,
)
reuse_decision = require_definition_action(
source,
principal=principal,
registry=registry,
action="derive",
)
source_revision = get_pipeline_revision(
session,
pipeline=source,
revision=payload.source_revision,
)
stored_tenant_id = None if payload.scope_type == "system" else tenant_id
scope_id = (
None
if payload.scope_type == "system"
else tenant_id
if payload.scope_type == "tenant"
else payload.scope_id
)
source_limits = _effective_governance_limits(
source,
decision_details=reuse_decision.details,
)
effective_limits = {
"inherit_to_lower_scopes": (
source_limits["inherit_to_lower_scopes"]
and payload.inherit_to_lower_scopes
),
"allow_run": source_limits["allow_run"] and payload.allow_run,
"allow_reuse": source_limits["allow_reuse"] and payload.allow_reuse,
"allow_automation": (
source_limits["allow_automation"]
and payload.allow_automation
),
}
provenance = {
"source_ref": f"pipeline:{source.id}",
"source_scope": {
"scope_type": source.scope_type,
"scope_id": source.scope_id,
},
"source_definition_kind": source.definition_kind,
"source_revision": source_revision.revision,
"source_hash": source_revision.content_hash,
"source_effective_limits": effective_limits,
"policy_decision": reuse_decision.to_dict(),
"derived_by": actor_id,
"derived_at": utcnow().isoformat(),
}
pipeline = DataflowPipeline(
tenant_id=stored_tenant_id,
scope_type=payload.scope_type,
scope_id=scope_id,
definition_kind=payload.definition_kind,
inherit_to_lower_scopes=effective_limits[
"inherit_to_lower_scopes"
],
allow_run=effective_limits["allow_run"],
allow_reuse=effective_limits["allow_reuse"],
allow_automation=effective_limits["allow_automation"],
derived_from_pipeline_id=source.id,
derived_from_revision=source_revision.revision,
derived_from_hash=source_revision.content_hash,
derivation_provenance=provenance,
name=payload.name.strip(),
description=_clean_optional(payload.description),
status="draft",
current_revision=1,
created_by=actor_id,
updated_by=actor_id,
metadata_={},
)
pipeline.revisions.append(
DataflowPipelineRevision(
tenant_id=stored_tenant_id,
revision=1,
schema_version=source_revision.schema_version,
graph=dict(source_revision.graph),
sql_text=source_revision.sql_text,
editor_mode=source_revision.editor_mode,
content_hash=source_revision.content_hash,
created_by=actor_id,
)
)
session.add(pipeline)
session.flush()
return pipeline
def delete_pipeline(
session: Session,
*,
@@ -228,7 +389,13 @@ def delete_pipeline(
return pipeline
def pipeline_response(session: Session, pipeline: DataflowPipeline) -> PipelineResponse:
def pipeline_response(
session: Session,
pipeline: DataflowPipeline,
*,
principal: ApiPrincipal,
registry: object | None,
) -> PipelineResponse:
revision = get_pipeline_revision(session, pipeline=pipeline)
return PipelineResponse(
id=pipeline.id,
@@ -242,6 +409,11 @@ def pipeline_response(session: Session, pipeline: DataflowPipeline) -> PipelineR
created_at=pipeline.created_at,
updated_at=pipeline.updated_at,
revision=PipelineRevisionResponse.model_validate(revision),
governance=definition_governance_payload(
pipeline,
principal=principal,
registry=registry,
),
)
@@ -359,6 +531,20 @@ def preview_pipeline(
revision: DataflowPipelineRevision | None = None
if payload.pipeline_id:
pipeline = get_pipeline(session, tenant_id=tenant_id, pipeline_id=payload.pipeline_id)
if principal is None:
raise DataflowConflictError(
"Saved pipeline previews require a tenant API principal."
)
action = "edit" if pipeline.status == "draft" else "run"
try:
require_definition_action(
pipeline,
principal=principal,
registry=registry,
action=action,
)
except PermissionError as exc:
raise DataflowConflictError(str(exc)) from exc
revision = get_pipeline_revision(session, pipeline=pipeline, revision=payload.revision)
graph = PipelineGraph.model_validate(revision.graph)
sql_text = revision.sql_text
@@ -567,6 +753,20 @@ def start_pipeline_run(
tenant_id=tenant_id,
pipeline_id=pipeline_id,
)
action = (
"run"
if request.invocation.kind in {"manual", "api", "backfill"}
else "automate"
)
try:
require_definition_action(
pipeline,
principal=principal,
registry=registry,
action=action,
)
except PermissionError as exc:
raise DataflowConflictError(str(exc)) from exc
revision = get_pipeline_revision(
session,
pipeline=pipeline,
@@ -610,6 +810,17 @@ def start_pipeline_run(
idempotency_key=idempotency_key,
request_hash=request_hash,
request_=_run_request_payload(request),
invocation_kind=request.invocation.kind,
trigger_id=_strip_ref(
request.invocation.trigger_ref or "",
"dataflow-trigger:",
),
trigger_delivery_id=_strip_ref(
request.invocation.delivery_ref or "",
"dataflow-trigger-delivery:",
),
correlation_id=request.invocation.correlation_id,
causation_id=request.invocation.causation_id,
source_fingerprints=[],
result_schema=[],
diagnostics=[],
@@ -766,6 +977,17 @@ def pipeline_run_response(
output_publication_ref=run.output_publication_ref,
output_datasource_ref=run.output_datasource_ref,
output_materialization_ref=run.output_materialization_ref,
invocation_kind=run.invocation_kind,
trigger_ref=(
f"dataflow-trigger:{run.trigger_id}" if run.trigger_id else None
),
delivery_ref=(
f"dataflow-trigger-delivery:{run.trigger_delivery_id}"
if run.trigger_delivery_id
else None
),
correlation_id=run.correlation_id,
causation_id=run.causation_id,
error=run.error,
started_at=run.started_at,
finished_at=run.finished_at,
@@ -796,6 +1018,15 @@ def pipeline_run_descriptor(
output_publication_ref=run.output_publication_ref,
output_datasource_ref=run.output_datasource_ref,
output_materialization_ref=run.output_materialization_ref,
invocation_kind=run.invocation_kind,
trigger_ref=(
f"dataflow-trigger:{run.trigger_id}" if run.trigger_id else None
),
delivery_ref=(
f"dataflow-trigger-delivery:{run.trigger_delivery_id}"
if run.trigger_delivery_id
else None
),
error=run.error,
started_at=run.started_at,
finished_at=run.finished_at,
@@ -1000,6 +1231,28 @@ def _run_request_payload(request: DataflowRunRequest) -> dict[str, object]:
if target
else None
),
"invocation": _invocation_payload(request.invocation),
}
def _invocation_payload(
invocation: AutomationInvocation,
) -> dict[str, object]:
return {
"kind": invocation.kind,
"trigger_ref": invocation.trigger_ref,
"delivery_ref": invocation.delivery_ref,
"event_id": invocation.event_id,
"event_type": invocation.event_type,
"correlation_id": invocation.correlation_id,
"causation_id": invocation.causation_id,
"scheduled_for": (
invocation.scheduled_for.isoformat()
if invocation.scheduled_for
else None
),
"requested_by": invocation.requested_by,
"metadata": dict(invocation.metadata),
}
@@ -1020,6 +1273,54 @@ def _clean_optional(value: str | None) -> str | None:
return cleaned or None
def _ancestor_governance_limits(
provenance: Mapping[str, object],
) -> dict[str, bool]:
raw = provenance.get("source_effective_limits")
limits = raw if isinstance(raw, Mapping) else {}
return {
key: value if isinstance((value := limits.get(key)), bool) else True
for key in (
"inherit_to_lower_scopes",
"allow_run",
"allow_reuse",
"allow_automation",
)
}
def _effective_governance_limits(
pipeline: DataflowPipeline,
*,
decision_details: Mapping[str, object] | None = None,
) -> dict[str, bool]:
ancestor = _ancestor_governance_limits(
pipeline.derivation_provenance
)
effective = {
"inherit_to_lower_scopes": (
pipeline.inherit_to_lower_scopes
and ancestor["inherit_to_lower_scopes"]
),
"allow_run": pipeline.allow_run and ancestor["allow_run"],
"allow_reuse": pipeline.allow_reuse and ancestor["allow_reuse"],
"allow_automation": (
pipeline.allow_automation and ancestor["allow_automation"]
),
}
policy_limits = (
decision_details.get("effective_limits")
if decision_details is not None
else None
)
if isinstance(policy_limits, Mapping):
for key in effective:
value = policy_limits.get(key)
if isinstance(value, bool):
effective[key] = effective[key] and value
return effective
__all__ = [
"DataflowConflictError",
"DataflowError",
@@ -1029,6 +1330,7 @@ __all__ = [
"cancel_pipeline_run",
"compile_sql_draft",
"create_pipeline",
"derive_pipeline",
"delete_pipeline",
"get_pipeline",
"get_pipeline_revision",

View File

@@ -0,0 +1,982 @@
from __future__ import annotations
from collections.abc import Mapping
from datetime import datetime, timedelta, timezone
import hashlib
import math
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal, has_scope
from govoplan_core.core.automation import (
AutomationInvocation,
AutomationPrincipalRequest,
automation_principal_provider,
)
from govoplan_core.core.dataflows import (
DataflowPublicationTarget,
DataflowRunRequest,
)
from govoplan_core.core.events import PlatformEvent
from govoplan_core.db.base import utcnow
from govoplan_dataflow.backend.db.models import (
DataflowPipeline,
DataflowPipelineRevision,
DataflowRun,
DataflowTrigger,
DataflowTriggerDelivery,
new_uuid,
)
from govoplan_dataflow.backend.governance import require_definition_action
from govoplan_dataflow.backend.schemas import (
DataflowEventDeliveryRequest,
DataflowTriggerCreateRequest,
DataflowTriggerDeliveryResponse,
DataflowTriggerDispatchResponse,
DataflowTriggerEvent,
DataflowTriggerResponse,
DataflowTriggerSchedule,
DataflowTriggerUpdateRequest,
PipelineGraph,
PipelineRunPublicationRequest,
)
from govoplan_dataflow.backend.service import (
DataflowConflictError,
DataflowNotFoundError,
get_pipeline,
get_pipeline_revision,
start_pipeline_run,
)
TRIGGER_READ_SCOPE = "dataflow:trigger:read"
TRIGGER_WRITE_SCOPE = "dataflow:trigger:write"
TRIGGER_DISPATCH_SCOPE = "dataflow:trigger:dispatch"
RUN_SCOPE = "dataflow:pipeline:run"
DATASOURCE_READ_SCOPE = "datasources:catalogue:read"
DATASOURCE_WRITE_SCOPE = "datasources:source:write"
def list_triggers(
session: Session,
*,
tenant_id: str,
pipeline_id: str,
) -> list[DataflowTrigger]:
get_pipeline(session, tenant_id=tenant_id, pipeline_id=pipeline_id)
return list(
session.scalars(
select(DataflowTrigger)
.where(
DataflowTrigger.tenant_id == tenant_id,
DataflowTrigger.pipeline_id == pipeline_id,
)
.order_by(DataflowTrigger.created_at, DataflowTrigger.id)
)
)
def get_trigger(
session: Session,
*,
tenant_id: str,
trigger_id: str,
) -> DataflowTrigger:
trigger = session.scalar(
select(DataflowTrigger).where(
DataflowTrigger.id == trigger_id,
DataflowTrigger.tenant_id == tenant_id,
)
)
if trigger is None:
raise DataflowNotFoundError("Dataflow trigger not found")
return trigger
def create_trigger(
session: Session,
*,
tenant_id: str,
pipeline_id: str,
actor_id: str | None,
principal: ApiPrincipal,
registry: object | None,
payload: DataflowTriggerCreateRequest,
) -> DataflowTrigger:
pipeline = get_pipeline(
session,
tenant_id=tenant_id,
pipeline_id=pipeline_id,
)
_require_automation(
pipeline,
principal=principal,
registry=registry,
require_provider=payload.enabled,
)
revision = get_pipeline_revision(
session,
pipeline=pipeline,
revision=payload.revision,
)
grant_scopes = required_trigger_scopes(
revision=revision,
publication=payload.publication,
)
_require_grant_scopes(principal, grant_scopes)
account_id, membership_id = _authorization_owner(principal)
trigger_id = new_uuid()
status = "enabled" if payload.enabled else "disabled"
trigger = DataflowTrigger(
id=trigger_id,
tenant_id=tenant_id,
pipeline_id=pipeline.id,
pipeline_revision_id=revision.id,
name=payload.name.strip(),
kind=payload.kind,
status=status,
revision=1,
config_=_trigger_config(payload),
publication_=(
payload.publication.model_dump(mode="json")
if payload.publication
else None
),
row_limit=payload.row_limit,
catch_up_policy=payload.catch_up_policy,
max_concurrent_runs=payload.max_concurrent_runs,
next_fire_at=(
_initial_fire(payload, now=utcnow())
if payload.enabled
else None
),
authorization_account_id=account_id,
authorization_membership_id=membership_id,
authorization_ref=f"dataflow-trigger:{trigger_id}",
grant_scopes=grant_scopes,
created_by=actor_id,
updated_by=actor_id,
)
session.add(trigger)
session.flush()
return trigger
def update_trigger(
session: Session,
*,
tenant_id: str,
trigger_id: str,
actor_id: str | None,
principal: ApiPrincipal,
registry: object | None,
payload: DataflowTriggerUpdateRequest,
) -> DataflowTrigger:
trigger = get_trigger(
session,
tenant_id=tenant_id,
trigger_id=trigger_id,
)
if payload.expected_revision != trigger.revision:
raise DataflowConflictError(
"Trigger changed on the server; reload before saving."
)
pipeline = get_pipeline(
session,
tenant_id=tenant_id,
pipeline_id=trigger.pipeline_id,
)
_require_automation(
pipeline,
principal=principal,
registry=registry,
require_provider=payload.enabled,
)
revision = get_pipeline_revision(
session,
pipeline=pipeline,
revision=payload.revision,
)
grant_scopes = required_trigger_scopes(
revision=revision,
publication=payload.publication,
)
_require_grant_scopes(principal, grant_scopes)
account_id, membership_id = _authorization_owner(principal)
trigger.pipeline_revision_id = revision.id
trigger.name = payload.name.strip()
trigger.kind = payload.kind
trigger.status = "enabled" if payload.enabled else "disabled"
trigger.revision += 1
trigger.config_ = _trigger_config(payload)
trigger.publication_ = (
payload.publication.model_dump(mode="json")
if payload.publication
else None
)
trigger.row_limit = payload.row_limit
trigger.catch_up_policy = payload.catch_up_policy
trigger.max_concurrent_runs = payload.max_concurrent_runs
trigger.next_fire_at = (
_initial_fire(payload, now=utcnow()) if payload.enabled else None
)
trigger.last_error = None
trigger.authorization_account_id = account_id
trigger.authorization_membership_id = membership_id
trigger.grant_scopes = grant_scopes
trigger.updated_by = actor_id
session.flush()
return trigger
def delete_trigger(
session: Session,
*,
tenant_id: str,
trigger_id: str,
) -> DataflowTrigger:
trigger = get_trigger(
session,
tenant_id=tenant_id,
trigger_id=trigger_id,
)
session.delete(trigger)
session.flush()
return trigger
def list_deliveries(
session: Session,
*,
tenant_id: str,
trigger_id: str,
limit: int = 50,
) -> list[DataflowTriggerDelivery]:
get_trigger(session, tenant_id=tenant_id, trigger_id=trigger_id)
return list(
session.scalars(
select(DataflowTriggerDelivery)
.where(
DataflowTriggerDelivery.tenant_id == tenant_id,
DataflowTriggerDelivery.trigger_id == trigger_id,
)
.order_by(
DataflowTriggerDelivery.created_at.desc(),
DataflowTriggerDelivery.id.desc(),
)
.limit(max(1, min(int(limit), 100)))
)
)
def ingest_event_delivery(
session: Session,
*,
tenant_id: str,
event: DataflowEventDeliveryRequest,
) -> int:
if event.classification not in {"public", "internal"}:
raise DataflowConflictError(
"Confidential and restricted events require the durable Core "
"event bridge before they can start Dataflows."
)
triggers = list(
session.scalars(
select(DataflowTrigger).where(
DataflowTrigger.tenant_id == tenant_id,
DataflowTrigger.kind == "event",
DataflowTrigger.status == "enabled",
)
)
)
queued = 0
for trigger in triggers:
configured = _event_config(trigger)
if not _event_matches(configured, event):
continue
delivery = DataflowTriggerDelivery(
tenant_id=tenant_id,
trigger_id=trigger.id,
pipeline_id=trigger.pipeline_id,
pipeline_revision_id=trigger.pipeline_revision_id,
source_key=f"event:{event.event_id}",
invocation_kind="event",
status="queued",
event_=event.model_dump(mode="json"),
)
try:
with session.begin_nested():
session.add(delivery)
session.flush()
except IntegrityError:
continue
queued += 1
session.flush()
return queued
def dispatch_due_triggers(
session: Session,
*,
registry: object | None,
now: datetime | None = None,
limit: int = 50,
tenant_id: str | None = None,
) -> DataflowTriggerDispatchResponse:
current = _as_utc(now or utcnow())
bounded_limit = max(1, min(int(limit), 200))
queued, schedule_skipped = _queue_due_schedules(
session,
now=current,
limit=bounded_limit,
tenant_id=tenant_id,
)
delivery_query = select(DataflowTriggerDelivery).where(
DataflowTriggerDelivery.status == "queued"
)
if tenant_id:
delivery_query = delivery_query.where(
DataflowTriggerDelivery.tenant_id == tenant_id
)
deliveries = list(
session.scalars(
delivery_query
.order_by(
DataflowTriggerDelivery.created_at,
DataflowTriggerDelivery.id,
)
.limit(bounded_limit)
.with_for_update(skip_locked=True)
)
)
counts = {
"processed": schedule_skipped,
"succeeded": 0,
"failed": 0,
"blocked": 0,
"skipped": schedule_skipped,
}
for delivery in deliveries:
result = _dispatch_delivery(
session,
delivery=delivery,
registry=registry,
now=current,
)
counts["processed"] += 1
counts[result] += 1
session.flush()
return DataflowTriggerDispatchResponse(queued=queued, **counts)
def required_trigger_scopes(
*,
revision: DataflowPipelineRevision,
publication: PipelineRunPublicationRequest | None,
) -> list[str]:
graph = PipelineGraph.model_validate(revision.graph)
scopes = {RUN_SCOPE}
if any(
node.type == "source.reference"
or bool(node.config.get("source_ref"))
for node in graph.nodes
):
scopes.add(DATASOURCE_READ_SCOPE)
if publication is not None:
scopes.add(DATASOURCE_WRITE_SCOPE)
return sorted(scopes)
def trigger_response(
session: Session,
trigger: DataflowTrigger,
) -> DataflowTriggerResponse:
revision = session.get(
DataflowPipelineRevision,
trigger.pipeline_revision_id,
)
if revision is None:
raise DataflowNotFoundError("Dataflow pipeline revision not found")
config = dict(trigger.config_)
return DataflowTriggerResponse(
id=trigger.id,
tenant_id=trigger.tenant_id,
pipeline_id=trigger.pipeline_id,
pipeline_revision=revision.revision,
name=trigger.name,
kind=trigger.kind, # type: ignore[arg-type]
status=trigger.status, # type: ignore[arg-type]
revision=trigger.revision,
schedule=(
DataflowTriggerSchedule.model_validate(config["schedule"])
if isinstance(config.get("schedule"), Mapping)
else None
),
event=(
DataflowTriggerEvent.model_validate(config["event"])
if isinstance(config.get("event"), Mapping)
else None
),
publication=(
PipelineRunPublicationRequest.model_validate(
trigger.publication_
)
if trigger.publication_
else None
),
row_limit=trigger.row_limit,
catch_up_policy=trigger.catch_up_policy, # type: ignore[arg-type]
max_concurrent_runs=trigger.max_concurrent_runs,
next_fire_at=trigger.next_fire_at,
last_fire_at=trigger.last_fire_at,
last_status=trigger.last_status,
last_error=trigger.last_error,
grant_scopes=list(trigger.grant_scopes),
authorization_provenance={
"authorization_ref": trigger.authorization_ref,
"account_id": trigger.authorization_account_id,
"membership_id": trigger.authorization_membership_id,
"grant_scopes": list(trigger.grant_scopes),
"resolution": "rechecked_at_each_delivery",
},
created_by=trigger.created_by,
updated_by=trigger.updated_by,
created_at=trigger.created_at,
updated_at=trigger.updated_at,
)
def delivery_response(
delivery: DataflowTriggerDelivery,
) -> DataflowTriggerDeliveryResponse:
event = delivery.event_ or {}
return DataflowTriggerDeliveryResponse(
id=delivery.id,
trigger_id=delivery.trigger_id,
invocation_kind=delivery.invocation_kind,
status=delivery.status, # type: ignore[arg-type]
scheduled_for=delivery.scheduled_for,
event_id=str(event.get("event_id")) if event.get("event_id") else None,
run_id=delivery.run_id,
attempts=delivery.attempts,
authorization_provenance=dict(
delivery.authorization_provenance
),
error=delivery.error,
created_at=delivery.created_at,
finished_at=delivery.finished_at,
)
def _require_automation(
pipeline: DataflowPipeline,
*,
principal: ApiPrincipal,
registry: object | None,
require_provider: bool,
) -> None:
action = "automate" if require_provider else "view"
try:
require_definition_action(
pipeline,
principal=principal,
registry=registry,
action=action,
)
except PermissionError as exc:
raise DataflowConflictError(str(exc)) from exc
if require_provider and automation_principal_provider(registry) is None:
raise DataflowConflictError(
"Automated Dataflows require the Access automation-principal "
"capability so authorization can be rechecked for every run."
)
def _authorization_owner(principal: ApiPrincipal) -> tuple[str, str]:
if not principal.account_id or not principal.membership_id:
raise DataflowConflictError(
"Automated Dataflows require a tenant account membership."
)
return principal.account_id, principal.membership_id
def _require_grant_scopes(
principal: ApiPrincipal,
scopes: list[str],
) -> None:
missing = [scope for scope in scopes if not has_scope(principal, scope)]
if missing:
raise DataflowConflictError(
"The trigger owner lacks required execution scopes: "
+ ", ".join(missing)
)
def _trigger_config(
payload: DataflowTriggerCreateRequest | DataflowTriggerUpdateRequest,
) -> dict[str, object]:
return {
"schedule": (
payload.schedule.model_dump(mode="json")
if payload.schedule
else None
),
"event": (
payload.event.model_dump(mode="json") if payload.event else None
),
}
def _initial_fire(
payload: DataflowTriggerCreateRequest | DataflowTriggerUpdateRequest,
*,
now: datetime,
) -> datetime | None:
if payload.kind == "event":
return None
schedule = payload.schedule
if schedule is None:
raise DataflowConflictError("Scheduled trigger configuration is missing.")
zone = _timezone(schedule.timezone)
if payload.kind == "once":
return _localized_utc(schedule.run_at, zone)
anchor = _localized_utc(schedule.anchor_at, zone) if schedule.anchor_at else now
interval = timedelta(seconds=int(schedule.interval_seconds or 0))
if interval.total_seconds() <= 0:
raise DataflowConflictError("Interval trigger duration is invalid.")
anchor = _as_utc(anchor)
current = _as_utc(now)
if anchor < current:
anchor += interval * math.ceil((current - anchor) / interval)
return anchor
def _timezone(name: str) -> ZoneInfo:
try:
return ZoneInfo(name)
except ZoneInfoNotFoundError as exc:
raise DataflowConflictError(f"Unknown timezone: {name}") from exc
def _localized_utc(
value: datetime | None,
zone: ZoneInfo,
) -> datetime:
if value is None:
raise DataflowConflictError("Scheduled trigger time is missing.")
localized = value.replace(tzinfo=zone) if value.tzinfo is None else value
return localized.astimezone(timezone.utc)
def _as_utc(value: datetime) -> datetime:
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc)
def _queue_due_schedules(
session: Session,
*,
now: datetime,
limit: int,
tenant_id: str | None,
) -> tuple[int, int]:
trigger_query = select(DataflowTrigger).where(
DataflowTrigger.status == "enabled",
DataflowTrigger.kind.in_(("once", "interval")),
DataflowTrigger.next_fire_at.is_not(None),
DataflowTrigger.next_fire_at <= now,
)
if tenant_id:
trigger_query = trigger_query.where(
DataflowTrigger.tenant_id == tenant_id
)
triggers = list(
session.scalars(
trigger_query
.order_by(
DataflowTrigger.next_fire_at,
DataflowTrigger.id,
)
.limit(limit)
.with_for_update(skip_locked=True)
)
)
queued = 0
skipped = 0
for trigger in triggers:
fire_at = _as_utc(trigger.next_fire_at or now)
source_key = (
"schedule:"
+ hashlib.sha256(
f"{trigger.id}:{fire_at.isoformat()}".encode()
).hexdigest()
)
missed = (
trigger.kind == "interval"
and trigger.catch_up_policy == "skip"
and _interval_was_missed(
trigger,
fire_at=fire_at,
now=now,
)
)
delivery = DataflowTriggerDelivery(
tenant_id=trigger.tenant_id,
trigger_id=trigger.id,
pipeline_id=trigger.pipeline_id,
pipeline_revision_id=trigger.pipeline_revision_id,
source_key=source_key,
invocation_kind="schedule",
status="skipped" if missed else "queued",
scheduled_for=fire_at,
error=(
"Scheduled occurrence was skipped by the trigger catch-up "
"policy."
if missed
else None
),
finished_at=now if missed else None,
)
try:
with session.begin_nested():
session.add(delivery)
session.flush()
except IntegrityError:
pass
else:
if missed:
skipped += 1
else:
queued += 1
trigger.last_fire_at = fire_at
trigger.last_status = "skipped" if missed else "queued"
trigger.last_error = delivery.error
if trigger.kind == "once":
trigger.status = "completed"
trigger.next_fire_at = None
else:
trigger.next_fire_at = _next_interval_fire(
trigger,
after=now,
)
session.flush()
return queued, skipped
def _interval_was_missed(
trigger: DataflowTrigger,
*,
fire_at: datetime,
now: datetime,
) -> bool:
schedule = _schedule_config(trigger)
interval = timedelta(seconds=int(schedule.interval_seconds or 0))
if interval.total_seconds() <= 0:
raise DataflowConflictError("Interval trigger duration is invalid.")
return fire_at + interval <= now
def _next_interval_fire(
trigger: DataflowTrigger,
*,
after: datetime,
) -> datetime:
schedule = _schedule_config(trigger)
seconds = int(schedule.interval_seconds or 0)
if seconds <= 0:
raise DataflowConflictError("Interval trigger duration is invalid.")
interval = timedelta(seconds=seconds)
next_fire = _as_utc(trigger.next_fire_at or after) + interval
current = _as_utc(after)
if next_fire <= current:
next_fire += interval * (
math.floor((current - next_fire) / interval) + 1
)
return next_fire
def _dispatch_delivery(
session: Session,
*,
delivery: DataflowTriggerDelivery,
registry: object | None,
now: datetime,
) -> str:
trigger = session.get(DataflowTrigger, delivery.trigger_id)
pipeline = session.get(DataflowPipeline, delivery.pipeline_id)
if trigger is None or pipeline is None:
return _finish_delivery(
delivery,
status="skipped",
now=now,
error="Trigger or pipeline no longer exists.",
)
if trigger.status not in {"enabled", "completed"}:
return _finish_delivery(
delivery,
status="skipped",
now=now,
error="Trigger is no longer enabled.",
)
if _running_count(session, trigger.id) >= trigger.max_concurrent_runs:
delivery.error = "Concurrency limit reached; delivery remains queued."
return "skipped"
delivery.status = "running"
delivery.claimed_at = now
delivery.attempts += 1
provider = automation_principal_provider(registry)
if provider is None:
return _finish_delivery(
delivery,
status="blocked",
now=now,
error="Access automation-principal capability is unavailable.",
)
resolution = provider.resolve_automation_principal(
session,
request=AutomationPrincipalRequest(
tenant_id=trigger.tenant_id,
account_id=trigger.authorization_account_id,
membership_id=trigger.authorization_membership_id,
authorization_ref=trigger.authorization_ref,
grant_scopes=tuple(trigger.grant_scopes),
context={
"trigger_ref": f"dataflow-trigger:{trigger.id}",
"pipeline_ref": f"pipeline:{trigger.pipeline_id}",
"delivery_ref": (
f"dataflow-trigger-delivery:{delivery.id}"
),
},
),
)
delivery.authorization_provenance = dict(resolution.provenance)
if not resolution.allowed or not isinstance(
resolution.principal,
ApiPrincipal,
):
return _finish_delivery(
delivery,
status="blocked",
now=now,
error=resolution.reason or "Automation authorization was denied.",
)
principal = resolution.principal
try:
require_definition_action(
pipeline,
principal=principal,
registry=registry,
action="automate",
)
publication = _publication_target(trigger)
event = delivery.event_ or {}
run, _ = start_pipeline_run(
session,
tenant_id=trigger.tenant_id,
actor_id=trigger.authorization_account_id,
principal=principal,
registry=registry,
request=DataflowRunRequest(
pipeline_ref=f"pipeline:{pipeline.id}",
revision=_revision_number(
session,
delivery.pipeline_revision_id,
),
idempotency_key=f"trigger-delivery:{delivery.id}",
row_limit=trigger.row_limit,
publication=publication,
invocation=AutomationInvocation(
kind=delivery.invocation_kind, # type: ignore[arg-type]
trigger_ref=f"dataflow-trigger:{trigger.id}",
delivery_ref=(
f"dataflow-trigger-delivery:{delivery.id}"
),
event_id=_optional_text(event.get("event_id")),
event_type=_optional_text(event.get("type")),
correlation_id=_optional_text(
event.get("correlation_id")
),
causation_id=_optional_text(event.get("causation_id")),
scheduled_for=delivery.scheduled_for,
requested_by=trigger.authorization_account_id,
),
),
)
except (DataflowConflictError, PermissionError, ValueError) as exc:
trigger.last_status = "blocked"
trigger.last_error = str(exc)
return _finish_delivery(
delivery,
status="blocked",
now=now,
error=str(exc),
)
delivery.run_id = run.id
trigger.last_status = run.status
trigger.last_error = run.error
status = "succeeded" if run.status == "succeeded" else "failed"
return _finish_delivery(
delivery,
status=status,
now=now,
error=run.error,
)
def _finish_delivery(
delivery: DataflowTriggerDelivery,
*,
status: str,
now: datetime,
error: str | None,
) -> str:
delivery.status = status
delivery.finished_at = now
delivery.error = error
return status
def _running_count(session: Session, trigger_id: str) -> int:
return int(
session.scalar(
select(func.count())
.select_from(DataflowRun)
.where(
DataflowRun.trigger_id == trigger_id,
DataflowRun.status.in_(("queued", "running")),
)
)
or 0
)
def _revision_number(session: Session, revision_id: str) -> int:
revision = session.get(DataflowPipelineRevision, revision_id)
if revision is None:
raise DataflowNotFoundError("Pinned pipeline revision no longer exists.")
return revision.revision
def _schedule_config(trigger: DataflowTrigger) -> DataflowTriggerSchedule:
value = trigger.config_.get("schedule")
if not isinstance(value, Mapping):
raise DataflowConflictError("Scheduled trigger configuration is missing.")
return DataflowTriggerSchedule.model_validate(value)
def _event_config(trigger: DataflowTrigger) -> DataflowTriggerEvent:
value = trigger.config_.get("event")
if not isinstance(value, Mapping):
raise DataflowConflictError("Event trigger configuration is missing.")
return DataflowTriggerEvent.model_validate(value)
def _event_matches(
configured: DataflowTriggerEvent,
event: DataflowEventDeliveryRequest,
) -> bool:
if configured.event_type != event.type:
return False
if configured.module_id and configured.module_id != event.module_id:
return False
return all(
event.payload.get(key) == value
for key, value in configured.filters.items()
)
def _publication_target(
trigger: DataflowTrigger,
) -> DataflowPublicationTarget | None:
if not trigger.publication_:
return None
value = PipelineRunPublicationRequest.model_validate(
trigger.publication_
)
return DataflowPublicationTarget(
target_datasource_ref=value.target_datasource_ref,
name=value.name,
source_name=value.source_name,
description=value.description,
freeze=value.freeze,
frozen_label=value.frozen_label,
set_current=value.set_current,
metadata=value.metadata,
)
def _optional_text(value: object) -> str | None:
if value is None:
return None
clean = str(value).strip()
return clean or None
class SqlDataflowTriggerDispatcher:
def __init__(self, *, registry: object | None = None) -> None:
self._registry = registry
def dispatch_due(
self,
session: object,
*,
now: datetime | None = None,
limit: int = 50,
) -> Mapping[str, object]:
if not isinstance(session, Session):
raise TypeError("Dataflow trigger dispatch requires a Session.")
return dispatch_due_triggers(
session,
registry=self._registry,
now=now,
limit=limit,
).model_dump(mode="json")
def ingest_event(
self,
session: object,
*,
event: PlatformEvent,
) -> Mapping[str, object]:
if not isinstance(session, Session):
raise TypeError("Dataflow event ingestion requires a Session.")
if event.tenant is None:
return {"queued": 0, "reason": "Event has no tenant context."}
queued = ingest_event_delivery(
session,
tenant_id=event.tenant.id,
event=DataflowEventDeliveryRequest(
event_id=event.event_id,
type=event.type,
module_id=event.module_id,
payload=dict(event.payload),
occurred_at=event.occurred_at,
correlation_id=event.correlation_id,
causation_id=event.causation_id,
classification=event.classification,
),
)
return {"queued": queued}
__all__ = [
"SqlDataflowTriggerDispatcher",
"create_trigger",
"delete_trigger",
"delivery_response",
"dispatch_due_triggers",
"get_trigger",
"ingest_event_delivery",
"list_deliveries",
"list_triggers",
"required_trigger_scopes",
"trigger_response",
"update_trigger",
]

View File

@@ -2,7 +2,10 @@ from __future__ import annotations
import unittest
from govoplan_core.core.dataflows import CAPABILITY_DATAFLOW_RUN_LIFECYCLE
from govoplan_core.core.dataflows import (
CAPABILITY_DATAFLOW_RUN_LIFECYCLE,
CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
)
from govoplan_core.core.modules import ModuleManifest
from govoplan_dataflow.backend.manifest import get_manifest
@@ -24,12 +27,17 @@ class DataflowManifestTests(unittest.TestCase):
CAPABILITY_DATAFLOW_RUN_LIFECYCLE,
manifest.capability_factories,
)
self.assertIn(
CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
manifest.capability_factories,
)
self.assertEqual(
{
"dataflow.pipeline_catalog",
"dataflow.pipeline_preview",
"dataflow.run_lifecycle",
"dataflow.dataset_output",
"dataflow.trigger_dispatcher",
},
{item.name for item in manifest.provides_interfaces},
)

View File

@@ -24,7 +24,7 @@ class DataflowMigrationTests(unittest.TestCase):
try:
with engine.connect() as connection:
self.assertIn(
"b8e3c6a1f4d9",
"e2f4a8c9d1b7",
set(MigrationContext.configure(connection).get_current_heads()),
)
self.assertEqual(
@@ -32,6 +32,8 @@ class DataflowMigrationTests(unittest.TestCase):
"dataflow_pipelines",
"dataflow_pipeline_revisions",
"dataflow_runs",
"dataflow_triggers",
"dataflow_trigger_deliveries",
},
{
name

View File

@@ -173,6 +173,7 @@ class DataflowServiceTests(unittest.TestCase):
payload=PipelineCreateRequest(
name="Monthly comparison",
description="First governed pipeline",
status="active",
graph=sample_graph(),
editor_mode="graph",
),
@@ -277,6 +278,7 @@ class DataflowServiceTests(unittest.TestCase):
tenant_id="tenant-1",
actor_id="user-1",
payload=PipelinePreviewRequest(pipeline_id=pipeline.id, row_limit=1),
principal=principal(),
)
self.session.commit()
@@ -316,6 +318,7 @@ class DataflowServiceTests(unittest.TestCase):
tenant_id="tenant-1",
actor_id="user-1",
payload=PipelinePreviewRequest(pipeline_id=pipeline.id),
principal=principal(),
)
self.session.commit()

527
tests/test_triggers.py Normal file
View File

@@ -0,0 +1,527 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import unittest
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker
from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.automation import AutomationPrincipalResolution
from govoplan_core.core.policy import PolicyDecision
from govoplan_core.db.base import Base, utcnow
from govoplan_dataflow.backend.db.models import (
DataflowPipeline,
DataflowPipelineRevision,
DataflowRun,
DataflowTrigger,
DataflowTriggerDelivery,
)
from govoplan_dataflow.backend.schemas import (
DataflowEventDeliveryRequest,
DataflowTriggerCreateRequest,
DataflowTriggerEvent,
DataflowTriggerSchedule,
PipelineCreateRequest,
PipelineDeriveRequest,
PipelineUpdateRequest,
)
from govoplan_dataflow.backend.service import (
DataflowConflictError,
create_pipeline,
derive_pipeline,
start_pipeline_run,
update_pipeline,
)
from govoplan_core.core.dataflows import DataflowRunRequest
from govoplan_dataflow.backend.triggers import (
create_trigger,
dispatch_due_triggers,
ingest_event_delivery,
)
POLICY_CAPABILITY = "policy.definitionGovernance"
AUTOMATION_CAPABILITY = "auth.automationPrincipalProvider"
def sample_graph():
from test_service import sample_graph as build_graph
return build_graph()
def principal() -> ApiPrincipal:
return ApiPrincipal(
principal=PrincipalRef(
account_id="account-1",
membership_id="membership-1",
tenant_id="tenant-1",
scopes=frozenset(
{
"dataflow:pipeline:run",
"dataflow:trigger:read",
"dataflow:trigger:write",
}
),
),
account=object(),
user=object(),
)
class AllowDefinitionPolicy:
def resolve_definition_action(self, *, request):
allowed = not (
request.definition_kind == "template"
and request.action in {"run", "automate"}
)
if request.action == "automate":
allowed = allowed and request.allow_automation
if request.action == "run":
allowed = (
allowed
and request.allow_run
and request.status == "active"
)
if request.action in {"reuse", "derive"}:
allowed = allowed and request.allow_reuse
return PolicyDecision(
allowed=allowed,
reason=None if allowed else "Definition action denied.",
)
class CurrentPrincipalProvider:
def __init__(self, actor: ApiPrincipal, *, allowed: bool = True) -> None:
self.actor = actor
self.allowed = allowed
def resolve_automation_principal(self, _session, *, request):
return AutomationPrincipalResolution(
allowed=self.allowed,
principal=self.actor if self.allowed else None,
reason=None if self.allowed else "Owner authorization revoked.",
granted_scopes=tuple(request.grant_scopes) if self.allowed else (),
missing_scopes=() if self.allowed else tuple(request.grant_scopes),
provenance={"status": "current" if self.allowed else "revoked"},
)
class Registry:
def __init__(
self,
actor: ApiPrincipal,
*,
automation_allowed: bool = True,
) -> None:
self.capabilities = {
POLICY_CAPABILITY: AllowDefinitionPolicy(),
AUTOMATION_CAPABILITY: CurrentPrincipalProvider(
actor,
allowed=automation_allowed,
),
}
def has_capability(self, name: str) -> bool:
return name in self.capabilities
def capability(self, name: str):
return self.capabilities[name]
class DataflowTriggerTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(
self.engine,
tables=[
DataflowPipeline.__table__,
DataflowPipelineRevision.__table__,
DataflowRun.__table__,
DataflowTrigger.__table__,
DataflowTriggerDelivery.__table__,
],
)
self.Session = sessionmaker(bind=self.engine)
self.session: Session = self.Session()
self.actor = principal()
self.registry = Registry(self.actor)
self.pipeline = create_pipeline(
self.session,
tenant_id="tenant-1",
actor_id="account-1",
payload=PipelineCreateRequest(
name="Automated comparison",
status="active",
allow_automation=True,
graph=sample_graph(),
),
)
self.session.commit()
def tearDown(self) -> None:
self.session.close()
Base.metadata.drop_all(
self.engine,
tables=[
DataflowTriggerDelivery.__table__,
DataflowTrigger.__table__,
DataflowRun.__table__,
DataflowPipelineRevision.__table__,
DataflowPipeline.__table__,
],
)
self.engine.dispose()
def _event_trigger(self) -> DataflowTrigger:
trigger = create_trigger(
self.session,
tenant_id="tenant-1",
pipeline_id=self.pipeline.id,
actor_id="account-1",
principal=self.actor,
registry=self.registry,
payload=DataflowTriggerCreateRequest(
name="Monthly file arrived",
kind="event",
enabled=True,
event=DataflowTriggerEvent(
event_type="files.upload.completed",
module_id="files",
filters={"folder_id": "incoming"},
),
),
)
self.session.commit()
return trigger
def test_interval_delivery_runs_pinned_revision_with_provenance(self) -> None:
trigger = create_trigger(
self.session,
tenant_id="tenant-1",
pipeline_id=self.pipeline.id,
actor_id="account-1",
principal=self.actor,
registry=self.registry,
payload=DataflowTriggerCreateRequest(
name="Hourly",
kind="interval",
enabled=True,
schedule=DataflowTriggerSchedule(
interval_seconds=3600,
timezone="Europe/Berlin",
),
),
)
now = utcnow()
trigger.next_fire_at = now - timedelta(minutes=1)
self.session.commit()
result = dispatch_due_triggers(
self.session,
registry=self.registry,
now=now,
)
self.session.commit()
delivery = self.session.scalar(
select(DataflowTriggerDelivery).where(
DataflowTriggerDelivery.trigger_id == trigger.id
)
)
run = self.session.get(DataflowRun, delivery.run_id)
self.assertEqual(1, result.queued)
self.assertEqual(1, result.succeeded)
self.assertEqual("succeeded", delivery.status)
self.assertEqual("schedule", run.invocation_kind)
self.assertEqual(trigger.id, run.trigger_id)
self.assertEqual(delivery.id, run.trigger_delivery_id)
self.assertEqual("current", delivery.authorization_provenance["status"])
def test_skip_catch_up_records_missed_interval_without_running(self) -> None:
trigger = create_trigger(
self.session,
tenant_id="tenant-1",
pipeline_id=self.pipeline.id,
actor_id="account-1",
principal=self.actor,
registry=self.registry,
payload=DataflowTriggerCreateRequest(
name="Skip stale intervals",
kind="interval",
enabled=True,
catch_up_policy="skip",
schedule=DataflowTriggerSchedule(
anchor_at=datetime(
2000,
1,
1,
tzinfo=timezone.utc,
),
interval_seconds=60,
),
),
)
now = utcnow()
trigger.next_fire_at = now - timedelta(minutes=3)
self.session.commit()
result = dispatch_due_triggers(
self.session,
registry=self.registry,
now=now,
)
self.session.commit()
delivery = self.session.scalar(
select(DataflowTriggerDelivery).where(
DataflowTriggerDelivery.trigger_id == trigger.id
)
)
self.assertEqual(0, result.queued)
self.assertEqual(1, result.skipped)
self.assertEqual("skipped", delivery.status)
self.assertIsNone(delivery.run_id)
next_fire_at = trigger.next_fire_at
self.assertIsNotNone(next_fire_at)
if next_fire_at.tzinfo is None:
next_fire_at = next_fire_at.replace(tzinfo=timezone.utc)
self.assertGreater(next_fire_at, now)
def test_disabled_trigger_can_be_prepared_for_a_draft(self) -> None:
self.pipeline.status = "draft"
self.session.flush()
trigger = create_trigger(
self.session,
tenant_id="tenant-1",
pipeline_id=self.pipeline.id,
actor_id="account-1",
principal=self.actor,
registry=self.registry,
payload=DataflowTriggerCreateRequest(
name="Prepared event",
kind="event",
enabled=False,
event=DataflowTriggerEvent(
event_type="files.upload.completed",
),
),
)
self.assertEqual("disabled", trigger.status)
self.assertIsNone(trigger.next_fire_at)
def test_event_ingress_is_filtered_and_idempotent(self) -> None:
trigger = self._event_trigger()
event = DataflowEventDeliveryRequest(
event_id="event-1",
type="files.upload.completed",
module_id="files",
payload={"folder_id": "incoming", "file_id": "file-1"},
occurred_at=utcnow(),
)
self.assertEqual(
1,
ingest_event_delivery(
self.session,
tenant_id="tenant-1",
event=event,
),
)
self.assertEqual(
0,
ingest_event_delivery(
self.session,
tenant_id="tenant-1",
event=event,
),
)
result = dispatch_due_triggers(
self.session,
registry=self.registry,
)
self.session.commit()
delivery = self.session.scalar(
select(DataflowTriggerDelivery).where(
DataflowTriggerDelivery.trigger_id == trigger.id
)
)
run = self.session.get(DataflowRun, delivery.run_id)
self.assertEqual(1, result.succeeded)
self.assertEqual("event", run.invocation_kind)
self.assertEqual("event-1", run.request_["invocation"]["event_id"])
def test_revoked_owner_blocks_delivery_before_execution(self) -> None:
trigger = self._event_trigger()
ingest_event_delivery(
self.session,
tenant_id="tenant-1",
event=DataflowEventDeliveryRequest(
event_id="event-revoked",
type="files.upload.completed",
module_id="files",
payload={"folder_id": "incoming"},
occurred_at=utcnow(),
),
)
result = dispatch_due_triggers(
self.session,
registry=Registry(self.actor, automation_allowed=False),
)
self.session.commit()
delivery = self.session.scalar(
select(DataflowTriggerDelivery).where(
DataflowTriggerDelivery.trigger_id == trigger.id
)
)
self.assertEqual(1, result.blocked)
self.assertEqual("blocked", delivery.status)
self.assertIn("revoked", delivery.error)
self.assertIsNone(delivery.run_id)
def test_restricted_event_requires_durable_bridge(self) -> None:
self._event_trigger()
with self.assertRaises(DataflowConflictError):
ingest_event_delivery(
self.session,
tenant_id="tenant-1",
event=DataflowEventDeliveryRequest(
event_id="event-secret",
type="files.upload.completed",
module_id="files",
payload={"folder_id": "incoming"},
occurred_at=utcnow(),
classification="restricted",
),
)
def test_template_derivation_pins_source_and_template_cannot_run(self) -> None:
template = create_pipeline(
self.session,
tenant_id="tenant-1",
actor_id="account-1",
payload=PipelineCreateRequest(
name="System import template",
graph=sample_graph(),
scope_type="system",
definition_kind="template",
inherit_to_lower_scopes=True,
allow_reuse=True,
),
)
derived = derive_pipeline(
self.session,
tenant_id="tenant-1",
actor_id="account-1",
principal=self.actor,
registry=self.registry,
source_pipeline_id=template.id,
payload=PipelineDeriveRequest(
name="Tenant monthly import",
definition_kind="flow",
allow_run=True,
),
)
self.session.commit()
self.assertEqual(template.id, derived.derived_from_pipeline_id)
self.assertEqual(1, derived.derived_from_revision)
self.assertEqual(
template.revisions[0].content_hash,
derived.derived_from_hash,
)
self.assertEqual(
"system",
derived.derivation_provenance["source_scope"]["scope_type"],
)
with self.assertRaises(DataflowConflictError):
start_pipeline_run(
self.session,
tenant_id="tenant-1",
actor_id="account-1",
principal=self.actor,
registry=self.registry,
request=DataflowRunRequest(
pipeline_ref=f"pipeline:{template.id}",
revision=1,
idempotency_key="template-must-not-run",
),
)
def test_derived_limits_cannot_be_broadened_transitively(self) -> None:
template = create_pipeline(
self.session,
tenant_id="tenant-1",
actor_id="account-1",
payload=PipelineCreateRequest(
name="Restricted template",
graph=sample_graph(),
definition_kind="template",
allow_reuse=True,
allow_automation=False,
inherit_to_lower_scopes=False,
),
)
derived = derive_pipeline(
self.session,
tenant_id="tenant-1",
actor_id="account-1",
principal=self.actor,
registry=self.registry,
source_pipeline_id=template.id,
payload=PipelineDeriveRequest(
name="Tenant flow",
allow_reuse=True,
allow_automation=True,
inherit_to_lower_scopes=True,
),
)
self.assertFalse(derived.allow_automation)
self.assertFalse(derived.inherit_to_lower_scopes)
update_pipeline(
self.session,
tenant_id="tenant-1",
pipeline_id=derived.id,
actor_id="account-1",
payload=PipelineUpdateRequest(
name=derived.name,
graph=sample_graph(),
status="draft",
expected_revision=1,
allow_reuse=True,
allow_automation=True,
inherit_to_lower_scopes=True,
),
)
grandchild = derive_pipeline(
self.session,
tenant_id="tenant-1",
actor_id="account-1",
principal=self.actor,
registry=self.registry,
source_pipeline_id=derived.id,
payload=PipelineDeriveRequest(
name="User flow",
scope_type="user",
scope_id="membership-1",
allow_automation=True,
inherit_to_lower_scopes=True,
),
)
self.assertFalse(derived.allow_automation)
self.assertFalse(derived.inherit_to_lower_scopes)
self.assertFalse(grandchild.allow_automation)
self.assertFalse(grandchild.inherit_to_lower_scopes)
if __name__ == "__main__":
unittest.main()

View File

@@ -2,6 +2,8 @@ import { apiFetch, type ApiSettings } from "@govoplan/core-webui";
import type { DefinitionGraphNodeType } from "@govoplan/core-webui/definition-graph";
export type PipelineStatus = "draft" | "active" | "archived";
export type DefinitionScopeType = "system" | "tenant" | "group" | "user";
export type DefinitionKind = "flow" | "template";
export type EditorMode = "graph" | "sql";
export type NodeCategory = "load" | "combine" | "filter" | "transform" | "output";
@@ -48,7 +50,7 @@ export type PipelineRevision = {
export type Pipeline = {
id: string;
tenant_id: string;
tenant_id: string | null;
name: string;
description?: string | null;
status: PipelineStatus;
@@ -58,6 +60,30 @@ export type Pipeline = {
created_at: string;
updated_at: string;
revision: PipelineRevision;
governance: PipelineGovernance;
};
export type DefinitionActionDecision = {
allowed: boolean;
reason?: string | null;
source_path: Array<Record<string, unknown>>;
requirements: string[];
details: Record<string, unknown>;
};
export type PipelineGovernance = {
scope_type: DefinitionScopeType;
scope_id?: string | null;
definition_kind: DefinitionKind;
inherit_to_lower_scopes: boolean;
allow_run: boolean;
allow_reuse: boolean;
allow_automation: boolean;
derived_from_pipeline_id?: string | null;
derived_from_revision?: number | null;
derived_from_hash?: string | null;
derivation_provenance: Record<string, unknown>;
actions: Record<string, DefinitionActionDecision>;
};
export type PipelinePayload = {
@@ -67,6 +93,13 @@ export type PipelinePayload = {
graph: PipelineGraph;
sql_text?: string | null;
editor_mode: EditorMode;
scope_type: DefinitionScopeType;
scope_id?: string | null;
definition_kind: DefinitionKind;
inherit_to_lower_scopes: boolean;
allow_run: boolean;
allow_reuse: boolean;
allow_automation: boolean;
};
export type PipelineValidation = {
@@ -179,6 +212,11 @@ export type PipelineRun = {
output_publication_ref?: string | null;
output_datasource_ref?: string | null;
output_materialization_ref?: string | null;
invocation_kind: string;
trigger_ref?: string | null;
delivery_ref?: string | null;
correlation_id?: string | null;
causation_id?: string | null;
error?: string | null;
started_at: string;
finished_at?: string | null;
@@ -187,6 +225,52 @@ export type PipelineRun = {
replayed: boolean;
};
export type DataflowTriggerKind = "once" | "interval" | "event";
export type DataflowTrigger = {
id: string;
tenant_id: string;
pipeline_id: string;
pipeline_revision: number;
name: string;
kind: DataflowTriggerKind;
status: "enabled" | "disabled" | "completed" | "blocked";
revision: number;
schedule?: {
run_at?: string | null;
anchor_at?: string | null;
interval_seconds?: number | null;
timezone: string;
} | null;
event?: {
event_type: string;
module_id?: string | null;
filters: Record<string, string | number | boolean | null>;
} | null;
row_limit: number;
catch_up_policy: "skip" | "coalesce";
max_concurrent_runs: number;
next_fire_at?: string | null;
last_fire_at?: string | null;
last_status?: string | null;
last_error?: string | null;
grant_scopes: string[];
authorization_provenance: Record<string, unknown>;
created_at: string;
updated_at: string;
};
export type DataflowTriggerPayload = {
name: string;
kind: DataflowTriggerKind;
revision?: number | null;
enabled: boolean;
schedule?: DataflowTrigger["schedule"];
event?: DataflowTrigger["event"];
row_limit: number;
catch_up_policy: "skip" | "coalesce";
max_concurrent_runs: number;
};
export async function listDataflowNodeTypes(settings: ApiSettings): Promise<NodeTypeDefinition[]> {
const response = await apiFetch<{ nodes: NodeTypeDefinition[] }>(
settings,
@@ -260,6 +344,75 @@ export function deleteDataflowPipeline(settings: ApiSettings, pipelineId: string
);
}
export function deriveDataflowPipeline(
settings: ApiSettings,
pipelineId: string,
payload: {
name: string;
description?: string | null;
source_revision?: number | null;
scope_type: DefinitionScopeType;
scope_id?: string | null;
definition_kind: DefinitionKind;
inherit_to_lower_scopes: boolean;
allow_run: boolean;
allow_reuse: boolean;
allow_automation: boolean;
}
): Promise<Pipeline> {
return apiFetch(
settings,
`/api/v1/dataflow/pipelines/${encodeURIComponent(pipelineId)}/derive`,
{ method: "POST", body: JSON.stringify(payload) }
);
}
export async function listDataflowTriggers(
settings: ApiSettings,
pipelineId: string
): Promise<DataflowTrigger[]> {
const response = await apiFetch<{ triggers: DataflowTrigger[] }>(
settings,
`/api/v1/dataflow/pipelines/${encodeURIComponent(pipelineId)}/triggers`
);
return response.triggers;
}
export function createDataflowTrigger(
settings: ApiSettings,
pipelineId: string,
payload: DataflowTriggerPayload
): Promise<DataflowTrigger> {
return apiFetch(
settings,
`/api/v1/dataflow/pipelines/${encodeURIComponent(pipelineId)}/triggers`,
{ method: "POST", body: JSON.stringify(payload) }
);
}
export function updateDataflowTrigger(
settings: ApiSettings,
triggerId: string,
payload: DataflowTriggerPayload & { expected_revision: number }
): Promise<DataflowTrigger> {
return apiFetch(
settings,
`/api/v1/dataflow/triggers/${encodeURIComponent(triggerId)}`,
{ method: "PUT", body: JSON.stringify(payload) }
);
}
export function deleteDataflowTrigger(
settings: ApiSettings,
triggerId: string
): Promise<{ deleted: boolean }> {
return apiFetch(
settings,
`/api/v1/dataflow/triggers/${encodeURIComponent(triggerId)}`,
{ method: "DELETE" }
);
}
export function validateDataflowPipeline(
settings: ApiSettings,
payload: { graph?: PipelineGraph; sql_text?: string; source_nodes?: PipelineGraphNode[] }

View File

@@ -7,7 +7,9 @@ import {
} from "react";
import {
CheckCircle2,
Clock3,
Code2,
CopyPlus,
DatabaseZap,
Network,
Play,
@@ -15,6 +17,7 @@ import {
RefreshCw,
RotateCcw,
Save,
Settings2,
Trash2,
TriangleAlert,
Upload
@@ -40,19 +43,27 @@ import {
import { ReactFlowProvider } from "@xyflow/react";
import {
compileDataflowSql,
createDataflowTrigger,
createDataflowPipeline,
createDataflowSourceSnapshot,
deleteDataflowPipeline,
deleteDataflowTrigger,
deriveDataflowPipeline,
listDataflowNodeTypes,
listDataflowPipelineRuns,
listDataflowPipelines,
listDataflowSources,
listDataflowTriggers,
previewDataflowPipeline,
runDataflowPipeline,
renderDataflowSql,
updateDataflowPipeline,
updateDataflowTrigger,
validateDataflowPipeline,
type DataflowDiagnostic,
type DataflowTrigger,
type DataflowTriggerKind,
type DataflowTriggerPayload,
type EditorMode,
type NodeCategory,
type NodeTypeDefinition,
@@ -102,20 +113,61 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
const [deleteOpen, setDeleteOpen] = useState(false);
const [snapshotOpen, setSnapshotOpen] = useState(false);
const [runOpen, setRunOpen] = useState(false);
const [definitionSettingsOpen, setDefinitionSettingsOpen] = useState(false);
const [deriveOpen, setDeriveOpen] = useState(false);
const [triggersOpen, setTriggersOpen] = useState(false);
const [nodeLibrary, setNodeLibrary] = useState<NodeTypeDefinition[]>(FALLBACK_NODE_LIBRARY);
const [sources, setSources] = useState<TabularSource[]>([]);
const [sourceCatalogueAvailable, setSourceCatalogueAvailable] = useState(false);
const [sourceCatalogueWritable, setSourceCatalogueWritable] = useState(false);
const dirty = Boolean(draft) && draftFingerprint(draft) !== draftFingerprint(savedDraft);
const canWrite = hasScope(auth, "dataflow:pipeline:write") || hasScope(auth, "dataflow:pipeline:admin");
const canRun = hasScope(auth, "dataflow:pipeline:run") || hasScope(auth, "dataflow:pipeline:admin");
const canEdit = canWrite && (
!draft?.id || draft.governance?.actions.edit?.allowed !== false
);
const canReuse = Boolean(
draft?.id
&& canWrite
&& draft.governance?.actions.derive?.allowed
);
const canStartSavedRun = Boolean(
draft?.id
&& draft.definitionKind === "flow"
&& draft.governance?.actions.run?.allowed
);
const canPreview = Boolean(
canRun
&& draft
&& (
!draft.id
|| dirty
|| draft.status === "draft"
? canEdit
: draft.governance?.actions.run?.allowed
)
);
const canManageTriggers = Boolean(
draft?.id
&& (
hasScope(auth, "dataflow:trigger:write")
|| hasScope(auth, "dataflow:pipeline:admin")
)
);
const canViewTriggers = Boolean(
draft?.id
&& (
hasScope(auth, "dataflow:trigger:read")
|| canManageTriggers
)
);
const canImportSources = canWrite
&& sourceCatalogueWritable
&& (
hasScope(auth, "datasources:stage:write")
|| hasScope(auth, "datasources:source:admin")
);
const dirty = Boolean(draft) && draftFingerprint(draft) !== draftFingerprint(savedDraft);
const selectedNode = useMemo(
() => draft?.graph.nodes.find((node) => node.id === selectedNodeId) ?? null,
[draft, selectedNodeId]
@@ -208,7 +260,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
}, [savedDraft]);
const saveDraft = useCallback(async (): Promise<boolean> => {
if (!draft || !canWrite || !draft.name.trim()) {
if (!draft || !canEdit || !draft.name.trim()) {
setError(!draft?.name.trim() ? "Pipeline name is required." : "You cannot save this pipeline.");
return false;
}
@@ -239,7 +291,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
} finally {
setSaving(false);
}
}, [canWrite, draft, settings]);
}, [canEdit, draft, settings]);
useUnsavedDraftGuard({
dirty,
@@ -425,7 +477,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
};
const addNode = (type: string) => {
if (!draft || !canWrite || uniqueNodeExists(draft, type)) return;
if (!draft || !canEdit || uniqueNodeExists(draft, type)) return;
const node = newNode(type, {
x: 120 + draft.graph.nodes.length * 70,
y: 100 + (draft.graph.nodes.length % 4) * 80
@@ -478,6 +530,9 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
<span>
<strong>{pipeline.name}</strong>
<small>Revision {pipeline.current_revision}</small>
<small>
{pipeline.governance.scope_type} · {pipeline.governance.definition_kind}
</small>
</span>
<StatusBadge status={pipeline.status} />
</button>
@@ -501,7 +556,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
value={draft.name}
onChange={(event) => updateDraft({ name: event.target.value })}
aria-label="Pipeline name"
disabled={!canWrite}
disabled={!canEdit}
/>
<input
className="dataflow-description-input"
@@ -509,16 +564,18 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
onChange={(event) => updateDraft({ description: event.target.value })}
placeholder="Description"
aria-label="Pipeline description"
disabled={!canWrite}
disabled={!canEdit}
/>
<select
value={draft.status}
onChange={(event) => updateDraft({ status: event.target.value as PipelineDraft["status"] })}
aria-label="Pipeline status"
disabled={!canWrite}
disabled={!canEdit}
>
<option value="draft">Draft</option>
<option value="active">Active</option>
<option value="active" disabled={draft.definitionKind === "template"}>
Active
</option>
<option value="archived">Archived</option>
</select>
</div>
@@ -536,16 +593,52 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
<Button onClick={() => void validate()} disabled={working}>
<CheckCircle2 size={16} /> Validate
</Button>
<Button variant="primary" onClick={() => void runPreview()} disabled={working || !canRun}>
<Button
variant="primary"
onClick={() => void runPreview()}
disabled={working || !canPreview}
disabledReason={
draft.governance?.actions.run?.reason ?? undefined
}
>
<Play size={16} /> Preview
</Button>
<Button
onClick={() => setRunOpen(true)}
disabled={working || !canRun || dirty || !draft.id || !draft.currentRevision}
title={dirty ? "Save the pipeline before starting a pinned run." : undefined}
disabled={working || !canRun || dirty || !canStartSavedRun || !draft.currentRevision}
disabledReason={
dirty
? "Save the pipeline before starting a pinned run."
: draft.governance?.actions.run?.reason ?? undefined
}
>
<DatabaseZap size={16} /> Run
</Button>
<IconButton
label="Definition settings"
icon={<Settings2 size={16} />}
variant="ghost"
onClick={() => setDefinitionSettingsOpen(true)}
disabled={!draft}
/>
{draft.id ? (
<IconButton
label="Reuse as scoped copy"
icon={<CopyPlus size={16} />}
variant="ghost"
onClick={() => setDeriveOpen(true)}
disabled={!canReuse}
/>
) : null}
{draft.id ? (
<IconButton
label="Automation triggers"
icon={<Clock3 size={16} />}
variant="ghost"
onClick={() => setTriggersOpen(true)}
disabled={!canViewTriggers}
/>
) : null}
<IconButton
label="Discard changes"
icon={<RotateCcw size={16} />}
@@ -559,13 +652,13 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
icon={<Trash2 size={16} />}
variant="danger"
onClick={() => setDeleteOpen(true)}
disabled={!canWrite || saving}
disabled={!canEdit || saving}
/>
) : null}
<Button
variant="primary"
onClick={() => void saveDraft()}
disabled={saving || working || !dirty || !canWrite}
disabled={saving || working || !dirty || !canEdit}
>
<Save size={16} /> {saving ? "Saving..." : "Save"}
</Button>
@@ -597,7 +690,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
<h3>{group.label}</h3>
{group.nodes.map((definition) => {
const Icon = dataflowNodeIcon(definition.icon);
const disabled = !canWrite || uniqueNodeExists(draft, definition.type);
const disabled = !canEdit || uniqueNodeExists(draft, definition.type);
return (
<button
key={definition.type}
@@ -628,7 +721,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
nodeDiagnostics={nodeDiagnostics}
nodeLibrary={nodeLibrary}
selectedNodeId={selectedNodeId}
readOnly={!canWrite}
readOnly={!canEdit}
onGraphChange={updateGraph}
onSelectNode={setSelectedNodeId}
/>
@@ -640,7 +733,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
<Button
variant="primary"
onClick={() => void applySql(false)}
disabled={working || !canWrite}
disabled={working || !canEdit}
>
<Code2 size={16} /> Apply SQL
</Button>
@@ -649,7 +742,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
value={draft.sqlText}
onChange={(event) => updateDraft({ sqlText: event.target.value })}
spellCheck={false}
disabled={!canWrite}
disabled={!canEdit}
aria-label="Dataflow SQL"
/>
</div>
@@ -660,7 +753,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
nodeLibrary={nodeLibrary}
sources={sources}
sourceCatalogueAvailable={sourceCatalogueAvailable}
readOnly={!canWrite}
readOnly={!canEdit}
onChange={(node: PipelineGraphNode) => updateGraph(updateGraphNode(draft.graph, node))}
onDelete={(nodeId) => {
updateGraph({
@@ -766,10 +859,587 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
}
}}
/>
<DefinitionSettingsDialog
open={definitionSettingsOpen}
draft={draft}
editable={canEdit}
onChange={updateDraft}
onClose={() => setDefinitionSettingsOpen(false)}
/>
<DerivePipelineDialog
open={deriveOpen}
settings={settings}
pipeline={draft?.id ? {
id: draft.id,
name: draft.name,
revision: draft.currentRevision ?? 1
} : null}
onClose={() => setDeriveOpen(false)}
onDerived={(pipeline) => {
const next = draftFromPipeline(pipeline);
setPipelines((current) => [
pipeline,
...current.filter((item) => item.id !== pipeline.id)
]);
setDraft(next);
setSavedDraft(structuredClone(next));
setSelectedNodeId(next.graph.nodes[0]?.id ?? null);
setDeriveOpen(false);
setSuccess("Created a pinned scoped copy.");
}}
/>
<DataflowTriggersDialog
open={triggersOpen}
settings={settings}
pipeline={draft?.id ? {
id: draft.id,
name: draft.name,
revision: draft.currentRevision ?? 1
} : null}
editable={canManageTriggers}
onClose={() => setTriggersOpen(false)}
/>
</main>
);
}
function DefinitionSettingsDialog({
open,
draft,
editable,
onChange,
onClose
}: {
open: boolean;
draft: PipelineDraft | null;
editable: boolean;
onChange: (patch: Partial<PipelineDraft>) => void;
onClose: () => void;
}) {
const provenance = draft?.governance?.actions.edit?.source_path ?? [];
return (
<Dialog
open={open}
title="Definition settings"
className="dataflow-definition-dialog"
onClose={onClose}
footer={<Button onClick={onClose}>Close</Button>}
>
{draft ? (
<div className="dataflow-definition-fields">
<FormField
label="Scope"
help="The scope determines ownership, visibility, and where Policy inheritance is resolved."
>
<select
value={draft.scopeType}
disabled={!editable || Boolean(draft.id)}
onChange={(event) => onChange({
scopeType: event.target.value as PipelineDraft["scopeType"],
scopeId: ""
})}
>
<option value="system">System</option>
<option value="tenant">Tenant</option>
<option value="group">Group</option>
<option value="user">User</option>
</select>
</FormField>
<FormField label="Scope ID">
<input
value={draft.scopeId}
disabled={
!editable
|| Boolean(draft.id)
|| draft.scopeType === "system"
|| draft.scopeType === "tenant"
}
placeholder={
draft.scopeType === "user"
? "Current user when empty"
: "Required for group"
}
onChange={(event) => onChange({ scopeId: event.target.value })}
/>
</FormField>
<FormField
label="Definition kind"
help="Templates can be reused or derived, but cannot be run or automated directly."
>
<select
value={draft.definitionKind}
disabled={!editable || Boolean(draft.id)}
onChange={(event) => onChange({
definitionKind: event.target.value as PipelineDraft["definitionKind"],
status: "draft"
})}
>
<option value="flow">Complete flow</option>
<option value="template">Template</option>
</select>
</FormField>
<div className="dataflow-definition-toggles">
<ToggleSwitch
label="Visible to lower scopes"
checked={draft.inheritToLowerScopes}
disabled={!editable}
onChange={(value) => onChange({ inheritToLowerScopes: value })}
/>
<ToggleSwitch
label="Allow runs"
checked={draft.allowRun}
disabled={!editable}
onChange={(value) => onChange({ allowRun: value })}
/>
<ToggleSwitch
label="Allow reuse"
checked={draft.allowReuse}
disabled={!editable}
onChange={(value) => onChange({ allowReuse: value })}
/>
<ToggleSwitch
label="Allow automation"
checked={draft.allowAutomation}
disabled={!editable}
onChange={(value) => onChange({ allowAutomation: value })}
/>
</div>
{draft.governance?.derived_from_pipeline_id ? (
<section className="dataflow-provenance">
<strong>Derived from</strong>
<span>
{draft.governance.derived_from_pipeline_id}
{" · revision "}
{draft.governance.derived_from_revision}
</span>
<code>{draft.governance.derived_from_hash}</code>
</section>
) : null}
{provenance.length ? (
<section className="dataflow-provenance">
<strong>Effective Policy path</strong>
{provenance.map((item, index) => (
<span key={`${String(item.path ?? item.scope_type)}-${index}`}>
{String(item.label ?? item.path ?? item.scope_type)}
</span>
))}
{!editable && draft.governance?.actions.edit?.reason ? (
<small>{draft.governance.actions.edit.reason}</small>
) : null}
</section>
) : null}
</div>
) : null}
</Dialog>
);
}
function DerivePipelineDialog({
open,
settings,
pipeline,
onClose,
onDerived
}: {
open: boolean;
settings: ApiSettings;
pipeline: { id: string; name: string; revision: number } | null;
onClose: () => void;
onDerived: (pipeline: Pipeline) => void;
}) {
const [name, setName] = useState("");
const [kind, setKind] = useState<"flow" | "template">("flow");
const [scopeType, setScopeType] = useState<"tenant" | "group" | "user">("tenant");
const [scopeId, setScopeId] = useState("");
const [busy, setBusy] = useState(false);
const [error, setError] = useState("");
useEffect(() => {
if (!open) return;
setName(`${pipeline?.name ?? "Pipeline"} copy`);
setKind("flow");
setScopeType("tenant");
setScopeId("");
setError("");
}, [open, pipeline?.id]);
const derive = async () => {
if (!pipeline || !name.trim()) return;
setBusy(true);
setError("");
try {
onDerived(await deriveDataflowPipeline(settings, pipeline.id, {
name: name.trim(),
source_revision: pipeline.revision,
scope_type: scopeType,
scope_id: scopeId.trim() || null,
definition_kind: kind,
inherit_to_lower_scopes: false,
allow_run: true,
allow_reuse: false,
allow_automation: false
}));
} catch (deriveError) {
setError(apiErrorMessage(deriveError));
} finally {
setBusy(false);
}
};
return (
<Dialog
open={open}
title="Reuse as scoped copy"
className="dataflow-definition-dialog"
closeDisabled={busy}
onClose={onClose}
footer={(
<>
<Button onClick={onClose} disabled={busy}>Cancel</Button>
<Button
variant="primary"
onClick={() => void derive()}
disabled={busy || !pipeline || !name.trim()}
>
<CopyPlus size={16} /> Create copy
</Button>
</>
)}
>
<div className="dataflow-definition-fields">
{error ? <DismissibleAlert tone="danger" resetKey={error}>{error}</DismissibleAlert> : null}
<FormField label="Name">
<input value={name} onChange={(event) => setName(event.target.value)} />
</FormField>
<FormField label="Target scope">
<select
value={scopeType}
onChange={(event) => setScopeType(event.target.value as typeof scopeType)}
>
<option value="tenant">Tenant</option>
<option value="group">Group</option>
<option value="user">User</option>
</select>
</FormField>
{scopeType !== "tenant" ? (
<FormField label="Scope ID">
<input
value={scopeId}
placeholder={scopeType === "user" ? "Current user when empty" : "Group ID"}
onChange={(event) => setScopeId(event.target.value)}
/>
</FormField>
) : null}
<FormField label="Copy kind">
<select value={kind} onChange={(event) => setKind(event.target.value as typeof kind)}>
<option value="flow">Complete flow</option>
<option value="template">Template</option>
</select>
</FormField>
<small>
The source revision and content hash are pinned. Source Policy limits
are retained and can only be narrowed in the copy.
</small>
</div>
</Dialog>
);
}
function DataflowTriggersDialog({
open,
settings,
pipeline,
editable,
onClose
}: {
open: boolean;
settings: ApiSettings;
pipeline: { id: string; name: string; revision: number } | null;
editable: boolean;
onClose: () => void;
}) {
const [triggers, setTriggers] = useState<DataflowTrigger[]>([]);
const [selectedId, setSelectedId] = useState<string | null>(null);
const [name, setName] = useState("");
const [kind, setKind] = useState<DataflowTriggerKind>("interval");
const [enabled, setEnabled] = useState(false);
const [runAt, setRunAt] = useState("");
const [intervalMinutes, setIntervalMinutes] = useState(60);
const [catchUpPolicy, setCatchUpPolicy] = useState<DataflowTrigger["catch_up_policy"]>("coalesce");
const [maxConcurrentRuns, setMaxConcurrentRuns] = useState(1);
const [eventType, setEventType] = useState("");
const [eventModule, setEventModule] = useState("");
const [eventFilters, setEventFilters] = useState("{}");
const [busy, setBusy] = useState(false);
const [error, setError] = useState("");
const selected = triggers.find((item) => item.id === selectedId) ?? null;
const load = useCallback(async () => {
if (!pipeline) return;
setBusy(true);
setError("");
try {
setTriggers(await listDataflowTriggers(settings, pipeline.id));
} catch (loadError) {
setError(apiErrorMessage(loadError));
} finally {
setBusy(false);
}
}, [pipeline?.id, settings]);
useEffect(() => {
if (!open || !pipeline) return;
setSelectedId(null);
setName(`${pipeline.name} schedule`);
setKind("interval");
setEnabled(false);
setRunAt("");
setIntervalMinutes(60);
setCatchUpPolicy("coalesce");
setMaxConcurrentRuns(1);
setEventType("");
setEventModule("");
setEventFilters("{}");
void load();
}, [open, pipeline?.id, load]);
const edit = (trigger: DataflowTrigger) => {
setSelectedId(trigger.id);
setName(trigger.name);
setKind(trigger.kind);
setEnabled(trigger.status === "enabled");
setRunAt(toLocalDateTime(trigger.schedule?.run_at ?? ""));
setIntervalMinutes(Math.max(1, Math.round((trigger.schedule?.interval_seconds ?? 3600) / 60)));
setCatchUpPolicy(trigger.catch_up_policy);
setMaxConcurrentRuns(trigger.max_concurrent_runs);
setEventType(trigger.event?.event_type ?? "");
setEventModule(trigger.event?.module_id ?? "");
setEventFilters(JSON.stringify(trigger.event?.filters ?? {}, null, 2));
setError("");
};
const reset = () => {
setSelectedId(null);
setName(`${pipeline?.name ?? "Pipeline"} schedule`);
setKind("interval");
setEnabled(false);
setRunAt("");
setIntervalMinutes(60);
setCatchUpPolicy("coalesce");
setMaxConcurrentRuns(1);
setEventType("");
setEventModule("");
setEventFilters("{}");
setError("");
};
const save = async () => {
if (!pipeline || !name.trim()) return;
setBusy(true);
setError("");
try {
const filters = kind === "event"
? JSON.parse(eventFilters || "{}") as Record<string, string | number | boolean | null>
: {};
const timezone = Intl.DateTimeFormat().resolvedOptions().timeZone || "UTC";
const payload: DataflowTriggerPayload = {
name: name.trim(),
kind,
revision: pipeline.revision,
enabled,
schedule: kind === "once"
? { run_at: new Date(runAt).toISOString(), timezone }
: kind === "interval"
? { interval_seconds: intervalMinutes * 60, timezone }
: null,
event: kind === "event" ? {
event_type: eventType.trim(),
module_id: eventModule.trim() || null,
filters
} : null,
row_limit: 500,
catch_up_policy: catchUpPolicy,
max_concurrent_runs: maxConcurrentRuns
};
const saved = selected
? await updateDataflowTrigger(settings, selected.id, {
...payload,
expected_revision: selected.revision
})
: await createDataflowTrigger(settings, pipeline.id, payload);
setTriggers((current) => [
saved,
...current.filter((item) => item.id !== saved.id)
]);
edit(saved);
} catch (saveError) {
setError(
saveError instanceof SyntaxError
? "Event filters must be a JSON object with scalar values."
: apiErrorMessage(saveError)
);
} finally {
setBusy(false);
}
};
const remove = async (trigger: DataflowTrigger) => {
setBusy(true);
setError("");
try {
await deleteDataflowTrigger(settings, trigger.id);
setTriggers((current) => current.filter((item) => item.id !== trigger.id));
reset();
} catch (deleteError) {
setError(apiErrorMessage(deleteError));
} finally {
setBusy(false);
}
};
const complete = Boolean(
name.trim()
&& (kind !== "once" || runAt)
&& (kind !== "event" || eventType.trim())
);
return (
<Dialog
open={open}
title={`Automation · ${pipeline?.name ?? "pipeline"}`}
className="dataflow-triggers-dialog"
closeDisabled={busy}
onClose={onClose}
footer={(
<>
<Button onClick={onClose} disabled={busy}>Close</Button>
<Button onClick={reset} disabled={busy || !editable}>New trigger</Button>
<Button variant="primary" onClick={() => void save()} disabled={busy || !editable || !complete}>
<Save size={16} /> Save trigger
</Button>
</>
)}
>
<div className="dataflow-triggers-layout">
<div className="dataflow-trigger-list">
{triggers.map((trigger) => (
<button
key={trigger.id}
type="button"
className={trigger.id === selectedId ? "is-selected" : ""}
onClick={() => edit(trigger)}
>
<span>
<strong>{trigger.name}</strong>
<small>
{trigger.kind}
{trigger.next_fire_at ? ` · ${new Date(trigger.next_fire_at).toLocaleString()}` : ""}
</small>
</span>
<StatusBadge status={trigger.status} label={trigger.status} />
</button>
))}
{!busy && !triggers.length ? <small>No triggers configured</small> : null}
</div>
<div className="dataflow-trigger-form">
{error ? <DismissibleAlert tone="danger" resetKey={error}>{error}</DismissibleAlert> : null}
<FormField label="Name">
<input disabled={!editable} value={name} onChange={(event) => setName(event.target.value)} />
</FormField>
<FormField label="Trigger">
<select disabled={!editable} value={kind} onChange={(event) => setKind(event.target.value as DataflowTriggerKind)}>
<option value="once">Given time</option>
<option value="interval">Interval</option>
<option value="event">Platform event</option>
</select>
</FormField>
{kind === "once" ? (
<FormField label="Run at">
<input disabled={!editable} type="datetime-local" value={runAt} onChange={(event) => setRunAt(event.target.value)} />
</FormField>
) : null}
{kind === "interval" ? (
<>
<FormField label="Interval (minutes)">
<input
type="number"
min={1}
disabled={!editable}
value={intervalMinutes}
onChange={(event) => setIntervalMinutes(Math.max(1, Number(event.target.value)))}
/>
</FormField>
<FormField label="Missed runs">
<select
disabled={!editable}
value={catchUpPolicy}
onChange={(event) => setCatchUpPolicy(event.target.value as DataflowTrigger["catch_up_policy"])}
>
<option value="coalesce">Run one latest occurrence</option>
<option value="skip">Skip missed occurrences</option>
</select>
</FormField>
</>
) : null}
{kind === "event" ? (
<>
<FormField label="Event type">
<input disabled={!editable} value={eventType} onChange={(event) => setEventType(event.target.value)} />
</FormField>
<FormField label="Source module">
<input disabled={!editable} value={eventModule} onChange={(event) => setEventModule(event.target.value)} />
</FormField>
<FormField
label="Exact payload filters"
help="Only direct scalar payload fields are matched. No expressions are executed."
>
<textarea disabled={!editable} value={eventFilters} onChange={(event) => setEventFilters(event.target.value)} />
</FormField>
</>
) : null}
<ToggleSwitch
label="Enabled"
checked={enabled}
disabled={busy || !editable}
onChange={setEnabled}
/>
<FormField label="Concurrent runs">
<input
type="number"
min={1}
max={10}
disabled={!editable}
value={maxConcurrentRuns}
onChange={(event) => {
const value = Number(event.target.value);
setMaxConcurrentRuns(Math.max(1, Math.min(10, value || 1)));
}}
/>
</FormField>
{selected ? (
<section className="dataflow-trigger-security">
<strong>Execution grant</strong>
{selected.grant_scopes.map((scope) => <code key={scope}>{scope}</code>)}
<small>Authorization is resolved again for every delivery.</small>
{selected.last_error ? <small className="is-error">{selected.last_error}</small> : null}
<Button variant="danger" onClick={() => void remove(selected)} disabled={busy || !editable}>
<Trash2 size={16} /> Delete trigger
</Button>
</section>
) : null}
</div>
</div>
</Dialog>
);
}
function toLocalDateTime(value: string): string {
if (!value) return "";
const date = new Date(value);
const offset = date.getTimezoneOffset() * 60_000;
return new Date(date.getTime() - offset).toISOString().slice(0, 16);
}
type RunMode = "run" | "publish";
function RunPipelineDialog({

View File

@@ -5,7 +5,10 @@ import type {
PipelineGraphNode,
PipelinePayload,
PipelineStatus,
NodeTypeDefinition
NodeTypeDefinition,
DefinitionKind,
DefinitionScopeType,
PipelineGovernance
} from "../../api/dataflow";
import { createDefinitionGraphNode } from "@govoplan/core-webui/definition-graph";
@@ -18,6 +21,14 @@ export type PipelineDraft = {
graph: PipelineGraph;
sqlText: string;
editorMode: EditorMode;
scopeType: DefinitionScopeType;
scopeId: string;
definitionKind: DefinitionKind;
inheritToLowerScopes: boolean;
allowRun: boolean;
allowReuse: boolean;
allowAutomation: boolean;
governance: PipelineGovernance | null;
};
const input = [{ id: "input", label: "Input", required: true, multiple: false, minimum_connections: 1 }];
@@ -154,7 +165,15 @@ export function draftFromPipeline(pipeline: Pipeline): PipelineDraft {
status: pipeline.status,
graph: structuredClone(pipeline.revision.graph),
sqlText: pipeline.revision.sql_text ?? "",
editorMode: pipeline.revision.editor_mode
editorMode: pipeline.revision.editor_mode,
scopeType: pipeline.governance.scope_type,
scopeId: pipeline.governance.scope_id ?? "",
definitionKind: pipeline.governance.definition_kind,
inheritToLowerScopes: pipeline.governance.inherit_to_lower_scopes,
allowRun: pipeline.governance.allow_run,
allowReuse: pipeline.governance.allow_reuse,
allowAutomation: pipeline.governance.allow_automation,
governance: pipeline.governance
};
}
@@ -167,6 +186,14 @@ export function sampleDraft(): PipelineDraft {
status: "draft",
editorMode: "graph",
sqlText: "",
scopeType: "tenant",
scopeId: "",
definitionKind: "flow",
inheritToLowerScopes: false,
allowRun: true,
allowReuse: false,
allowAutomation: false,
governance: null,
graph: {
schema_version: 1,
nodes: [
@@ -238,7 +265,14 @@ export function pipelinePayload(draft: PipelineDraft): PipelinePayload {
status: draft.status,
graph: draft.graph,
sql_text: draft.sqlText.trim() || null,
editor_mode: draft.editorMode
editor_mode: draft.editorMode,
scope_type: draft.scopeType,
scope_id: draft.scopeId.trim() || null,
definition_kind: draft.definitionKind,
inherit_to_lower_scopes: draft.inheritToLowerScopes,
allow_run: draft.allowRun,
allow_reuse: draft.allowReuse,
allow_automation: draft.allowAutomation
};
}
@@ -254,7 +288,14 @@ export function draftFingerprint(draft: PipelineDraft | null): string {
status: draft.status,
graph: draft.graph,
sqlText: draft.sqlText,
editorMode: draft.editorMode
editorMode: draft.editorMode,
scopeType: draft.scopeType,
scopeId: draft.scopeId,
definitionKind: draft.definitionKind,
inheritToLowerScopes: draft.inheritToLowerScopes,
allowRun: draft.allowRun,
allowReuse: draft.allowReuse,
allowAutomation: draft.allowAutomation
});
}

View File

@@ -133,6 +133,128 @@
white-space: nowrap;
}
.dataflow-definition-dialog {
width: min(620px, calc(100vw - 32px));
}
.dataflow-definition-fields,
.dataflow-trigger-form {
display: grid;
gap: 12px;
}
.dataflow-definition-fields select,
.dataflow-definition-fields input,
.dataflow-trigger-form select,
.dataflow-trigger-form input,
.dataflow-trigger-form textarea {
width: 100%;
}
.dataflow-definition-toggles {
display: grid;
grid-template-columns: repeat(2, minmax(0, 1fr));
gap: 10px 18px;
}
.dataflow-provenance,
.dataflow-trigger-security {
display: grid;
gap: 6px;
padding: 10px;
border: var(--border-line);
border-radius: var(--radius-sm);
background: var(--panel-soft);
}
.dataflow-provenance span,
.dataflow-provenance small,
.dataflow-trigger-security small {
color: var(--muted);
font-size: 12px;
}
.dataflow-provenance code,
.dataflow-trigger-security code {
overflow: hidden;
color: var(--text);
font-size: 11px;
text-overflow: ellipsis;
}
.dataflow-triggers-dialog {
width: min(920px, calc(100vw - 32px));
}
.dataflow-triggers-layout {
display: grid;
grid-template-columns: minmax(230px, 0.8fr) minmax(320px, 1.2fr);
gap: 14px;
min-height: 420px;
}
.dataflow-trigger-list {
min-width: 0;
overflow: auto;
border-right: var(--border-line);
padding-right: 10px;
}
.dataflow-trigger-list > button {
display: grid;
grid-template-columns: minmax(0, 1fr) auto;
align-items: center;
gap: 8px;
width: 100%;
min-height: 52px;
padding: 8px;
border: 0;
border-radius: var(--radius-sm);
background: transparent;
color: var(--text);
text-align: left;
}
.dataflow-trigger-list > button:hover,
.dataflow-trigger-list > button.is-selected {
background: var(--primary-soft);
}
.dataflow-trigger-list span,
.dataflow-trigger-list strong,
.dataflow-trigger-list small {
display: block;
min-width: 0;
}
.dataflow-trigger-list small {
margin-top: 4px;
color: var(--muted);
}
.dataflow-trigger-form textarea {
min-height: 88px;
resize: vertical;
}
.dataflow-trigger-security .is-error {
color: var(--danger);
}
@media (max-width: 760px) {
.dataflow-triggers-layout,
.dataflow-definition-toggles {
grid-template-columns: 1fr;
}
.dataflow-trigger-list {
max-height: 180px;
border-right: 0;
border-bottom: var(--border-line);
padding: 0 0 10px;
}
}
.dataflow-pipeline-list strong {
color: var(--text-strong);
font-size: 13px;