diff --git a/README.md b/README.md index d1255c8..bb70617 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/src/govoplan_dataflow/backend/db/models.py b/src/govoplan_dataflow/backend/db/models.py index a79d978..5037ec3 100644 --- a/src/govoplan_dataflow/backend/db/models.py +++ b/src/govoplan_dataflow/backend/db/models.py @@ -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", ] diff --git a/src/govoplan_dataflow/backend/governance.py b/src/govoplan_dataflow/backend/governance.py new file mode 100644 index 0000000..09cabef --- /dev/null +++ b/src/govoplan_dataflow/backend/governance.py @@ -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", +] diff --git a/src/govoplan_dataflow/backend/manifest.py b/src/govoplan_dataflow/backend/manifest.py index 143a2aa..73336c3 100644 --- a/src/govoplan_dataflow/backend/manifest.py +++ b/src/govoplan_dataflow/backend/manifest.py @@ -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", diff --git a/src/govoplan_dataflow/backend/migrations/versions/e2f4a8c9d1b7_v0114_governed_triggers.py b/src/govoplan_dataflow/backend/migrations/versions/e2f4a8c9d1b7_v0114_governed_triggers.py new file mode 100644 index 0000000..f8839dc --- /dev/null +++ b/src/govoplan_dataflow/backend/migrations/versions/e2f4a8c9d1b7_v0114_governed_triggers.py @@ -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, + ) diff --git a/src/govoplan_dataflow/backend/router.py b/src/govoplan_dataflow/backend/router.py index e6500b7..77ce783 100644 --- a/src/govoplan_dataflow/backend/router.py +++ b/src/govoplan_dataflow/backend/router.py @@ -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, }, diff --git a/src/govoplan_dataflow/backend/schemas.py b/src/govoplan_dataflow/backend/schemas.py index 03fc678..f3e4df2 100644 --- a/src/govoplan_dataflow/backend/schemas.py +++ b/src/govoplan_dataflow/backend/schemas.py @@ -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 diff --git a/src/govoplan_dataflow/backend/service.py b/src/govoplan_dataflow/backend/service.py index ad6dd89..605e293 100644 --- a/src/govoplan_dataflow/backend/service.py +++ b/src/govoplan_dataflow/backend/service.py @@ -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", diff --git a/src/govoplan_dataflow/backend/triggers.py b/src/govoplan_dataflow/backend/triggers.py new file mode 100644 index 0000000..111ebc7 --- /dev/null +++ b/src/govoplan_dataflow/backend/triggers.py @@ -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", +] diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 13c50ba..162f4d7 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -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}, ) diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 37bc208..01d21f8 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -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 diff --git a/tests/test_service.py b/tests/test_service.py index 3440691..e46dc60 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -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() diff --git a/tests/test_triggers.py b/tests/test_triggers.py new file mode 100644 index 0000000..db91ca2 --- /dev/null +++ b/tests/test_triggers.py @@ -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() diff --git a/webui/src/api/dataflow.ts b/webui/src/api/dataflow.ts index 55cf9e7..36c7542 100644 --- a/webui/src/api/dataflow.ts +++ b/webui/src/api/dataflow.ts @@ -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>; + requirements: string[]; + details: Record; +}; + +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; + actions: Record; }; 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; + } | 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; + 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 { 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 { + 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 { + 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 { + 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 { + 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[] } diff --git a/webui/src/features/dataflow/DataflowPage.tsx b/webui/src/features/dataflow/DataflowPage.tsx index 431ae5e..b378dfc 100644 --- a/webui/src/features/dataflow/DataflowPage.tsx +++ b/webui/src/features/dataflow/DataflowPage.tsx @@ -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(FALLBACK_NODE_LIBRARY); const [sources, setSources] = useState([]); 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 => { - 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 {pipeline.name} Revision {pipeline.current_revision} + + {pipeline.governance.scope_type} · {pipeline.governance.definition_kind} + @@ -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} /> updateDraft({ description: event.target.value })} placeholder="Description" aria-label="Pipeline description" - disabled={!canWrite} + disabled={!canEdit} /> @@ -536,16 +593,52 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings - + } + variant="ghost" + onClick={() => setDefinitionSettingsOpen(true)} + disabled={!draft} + /> + {draft.id ? ( + } + variant="ghost" + onClick={() => setDeriveOpen(true)} + disabled={!canReuse} + /> + ) : null} + {draft.id ? ( + } + variant="ghost" + onClick={() => setTriggersOpen(true)} + disabled={!canViewTriggers} + /> + ) : null} } @@ -559,13 +652,13 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings icon={} variant="danger" onClick={() => setDeleteOpen(true)} - disabled={!canWrite || saving} + disabled={!canEdit || saving} /> ) : null} @@ -597,7 +690,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings

{group.label}

{group.nodes.map((definition) => { const Icon = dataflowNodeIcon(definition.icon); - const disabled = !canWrite || uniqueNodeExists(draft, definition.type); + const disabled = !canEdit || uniqueNodeExists(draft, definition.type); return ( @@ -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" /> @@ -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 } }} /> + setDefinitionSettingsOpen(false)} + /> + 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."); + }} + /> + setTriggersOpen(false)} + /> ); } +function DefinitionSettingsDialog({ + open, + draft, + editable, + onChange, + onClose +}: { + open: boolean; + draft: PipelineDraft | null; + editable: boolean; + onChange: (patch: Partial) => void; + onClose: () => void; +}) { + const provenance = draft?.governance?.actions.edit?.source_path ?? []; + return ( + Close} + > + {draft ? ( +
+ + + + + onChange({ scopeId: event.target.value })} + /> + + + + +
+ onChange({ inheritToLowerScopes: value })} + /> + onChange({ allowRun: value })} + /> + onChange({ allowReuse: value })} + /> + onChange({ allowAutomation: value })} + /> +
+ {draft.governance?.derived_from_pipeline_id ? ( +
+ Derived from + + {draft.governance.derived_from_pipeline_id} + {" · revision "} + {draft.governance.derived_from_revision} + + {draft.governance.derived_from_hash} +
+ ) : null} + {provenance.length ? ( +
+ Effective Policy path + {provenance.map((item, index) => ( + + {String(item.label ?? item.path ?? item.scope_type)} + + ))} + {!editable && draft.governance?.actions.edit?.reason ? ( + {draft.governance.actions.edit.reason} + ) : null} +
+ ) : null} +
+ ) : null} + + ); +} + +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 ( + + + + + )} + > +
+ {error ? {error} : null} + + setName(event.target.value)} /> + + + + + {scopeType !== "tenant" ? ( + + setScopeId(event.target.value)} + /> + + ) : null} + + + + + The source revision and content hash are pinned. Source Policy limits + are retained and can only be narrowed in the copy. + +
+ + ); +} + +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([]); + const [selectedId, setSelectedId] = useState(null); + const [name, setName] = useState(""); + const [kind, setKind] = useState("interval"); + const [enabled, setEnabled] = useState(false); + const [runAt, setRunAt] = useState(""); + const [intervalMinutes, setIntervalMinutes] = useState(60); + const [catchUpPolicy, setCatchUpPolicy] = useState("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 + : {}; + 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 ( + + + + + + )} + > +
+
+ {triggers.map((trigger) => ( + + ))} + {!busy && !triggers.length ? No triggers configured : null} +
+
+ {error ? {error} : null} + + setName(event.target.value)} /> + + + + + {kind === "once" ? ( + + setRunAt(event.target.value)} /> + + ) : null} + {kind === "interval" ? ( + <> + + setIntervalMinutes(Math.max(1, Number(event.target.value)))} + /> + + + + + + ) : null} + {kind === "event" ? ( + <> + + setEventType(event.target.value)} /> + + + setEventModule(event.target.value)} /> + + +