diff --git a/README.md b/README.md index 195e336..d1255c8 100644 --- a/README.md +++ b/README.md @@ -2,9 +2,9 @@ GovOPlaN Dataflow defines and runs governed tabular transformation pipelines. Power users can work with the same pipeline as a graphical node graph or as a -constrained SQL query. Every saved definition change produces an immutable revision, and -every preview records diagnostics and reproducibility metadata without storing -the previewed row contents. +constrained SQL query. Every saved definition change produces an immutable +revision, and every preview records diagnostics and reproducibility metadata +without storing the previewed row contents. ## Boundary @@ -55,6 +55,14 @@ result-byte, graph-node, and response-row bounds. Saved previews record the pipeline revision, executor version, source fingerprints, node diagnostics, and output summary, but not source or result rows. +Saved revisions can also be started through the versioned +`dataflow.runLifecycle` capability or the Run dialog. The first runner is +synchronous and bounded. It records an idempotent terminal run, pins the +pipeline revision, and can atomically publish a complete result through +`datasources.publication`. Publication is rejected when a source or result was +truncated; larger asynchronous and artifact-backed execution belongs to a +subsequent runner slice. + ## Development ```bash diff --git a/src/govoplan_dataflow/backend/db/models.py b/src/govoplan_dataflow/backend/db/models.py index a307cb8..a79d978 100644 --- a/src/govoplan_dataflow/backend/db/models.py +++ b/src/govoplan_dataflow/backend/db/models.py @@ -73,6 +73,12 @@ class DataflowPipelineRevision(Base, TimestampMixin): class DataflowRun(Base, TimestampMixin): __tablename__ = "dataflow_runs" __table_args__ = ( + UniqueConstraint( + "tenant_id", + "pipeline_id", + "idempotency_key", + name="uq_dataflow_run_idempotency", + ), Index("ix_dataflow_runs_tenant_status", "tenant_id", "status"), Index("ix_dataflow_runs_pipeline_created", "pipeline_id", "created_at"), ) @@ -93,11 +99,35 @@ class DataflowRun(Base, TimestampMixin): status: Mapped[str] = mapped_column(String(30), nullable=False, index=True) executor_version: Mapped[str] = mapped_column(String(40), nullable=False) definition_hash: Mapped[str] = mapped_column(String(64), nullable=False) + idempotency_key: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + index=True, + ) + request_hash: Mapped[str | None] = mapped_column(String(64), nullable=True) + request_: Mapped[dict[str, Any]] = mapped_column( + "request", + JSON, + default=dict, + nullable=False, + ) 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) input_row_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) output_row_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + output_publication_ref: Mapped[str | None] = mapped_column( + String(500), + nullable=True, + ) + output_datasource_ref: Mapped[str | None] = mapped_column( + String(500), + nullable=True, + ) + output_materialization_ref: Mapped[str | None] = mapped_column( + String(500), + nullable=True, + ) started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) error: Mapped[str | None] = mapped_column(Text) diff --git a/src/govoplan_dataflow/backend/manifest.py b/src/govoplan_dataflow/backend/manifest.py index e0903ee..143a2aa 100644 --- a/src/govoplan_dataflow/backend/manifest.py +++ b/src/govoplan_dataflow/backend/manifest.py @@ -6,6 +6,7 @@ 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.modules import ( DocumentationTopic, FrontendModule, @@ -21,6 +22,7 @@ from govoplan_core.core.modules import ( from govoplan_core.core.datasources import ( CAPABILITY_DATASOURCE_CATALOGUE, CAPABILITY_DATASOURCE_LIFECYCLE, + CAPABILITY_DATASOURCE_PUBLICATION, ) from govoplan_core.db.base import Base from govoplan_dataflow.backend.db import models as dataflow_models @@ -143,6 +145,12 @@ def _dataflow_router(context: ModuleContext): return router +def _run_provider(context: ModuleContext): + from govoplan_dataflow.backend.service import SqlDataflowRunLifecycleProvider + + return SqlDataflowRunLifecycleProvider(registry=context.registry) + + def _tenant_summary(session, tenant_id: str) -> dict[str, int]: return { "dataflow_pipelines": ( @@ -180,6 +188,7 @@ manifest = ModuleManifest( optional_capabilities=( CAPABILITY_DATASOURCE_CATALOGUE, CAPABILITY_DATASOURCE_LIFECYCLE, + CAPABILITY_DATASOURCE_PUBLICATION, ), provides_interfaces=( ModuleInterfaceProvider(name="dataflow.pipeline_catalog", version=MODULE_VERSION), @@ -200,6 +209,12 @@ manifest = ModuleManifest( version_max_exclusive="1.0.0", optional=True, ), + ModuleInterfaceRequirement( + name="datasources.publication", + version_min="0.1.0", + version_max_exclusive="1.0.0", + optional=True, + ), ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, @@ -226,6 +241,9 @@ manifest = ModuleManifest( ), ), route_factory=_dataflow_router, + capability_factories={ + CAPABILITY_DATAFLOW_RUN_LIFECYCLE: _run_provider, + }, tenant_summary_providers=(_tenant_summary,), migration_spec=MigrationSpec( module_id=MODULE_ID, diff --git a/src/govoplan_dataflow/backend/migrations/versions/b8e3c6a1f4d9_v0114_dataflow_runs.py b/src/govoplan_dataflow/backend/migrations/versions/b8e3c6a1f4d9_v0114_dataflow_runs.py new file mode 100644 index 0000000..906470b --- /dev/null +++ b/src/govoplan_dataflow/backend/migrations/versions/b8e3c6a1f4d9_v0114_dataflow_runs.py @@ -0,0 +1,83 @@ +"""v0.1.14 Dataflow run lifecycle + +Revision ID: b8e3c6a1f4d9 +Revises: d4f7a1c8e2b0 +Create Date: 2026-07-28 00:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "b8e3c6a1f4d9" +down_revision = "d4f7a1c8e2b0" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("dataflow_runs") as batch_op: + batch_op.add_column( + sa.Column("idempotency_key", sa.String(length=255), nullable=True) + ) + batch_op.add_column( + sa.Column("request_hash", sa.String(length=64), nullable=True) + ) + batch_op.add_column( + sa.Column( + "request", + sa.JSON(), + nullable=False, + server_default=sa.text("'{}'"), + ) + ) + batch_op.add_column( + sa.Column( + "output_publication_ref", + sa.String(length=500), + nullable=True, + ) + ) + batch_op.add_column( + sa.Column( + "output_datasource_ref", + sa.String(length=500), + nullable=True, + ) + ) + batch_op.add_column( + sa.Column( + "output_materialization_ref", + sa.String(length=500), + nullable=True, + ) + ) + batch_op.create_unique_constraint( + "uq_dataflow_run_idempotency", + ["tenant_id", "pipeline_id", "idempotency_key"], + ) + op.create_index( + op.f("ix_dataflow_runs_idempotency_key"), + "dataflow_runs", + ["idempotency_key"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_index( + op.f("ix_dataflow_runs_idempotency_key"), + table_name="dataflow_runs", + ) + with op.batch_alter_table("dataflow_runs") as batch_op: + batch_op.drop_constraint( + "uq_dataflow_run_idempotency", + type_="unique", + ) + batch_op.drop_column("output_materialization_ref") + batch_op.drop_column("output_datasource_ref") + batch_op.drop_column("output_publication_ref") + batch_op.drop_column("request") + batch_op.drop_column("request_hash") + batch_op.drop_column("idempotency_key") diff --git a/src/govoplan_dataflow/backend/router.py b/src/govoplan_dataflow/backend/router.py index f3d65c0..e6500b7 100644 --- a/src/govoplan_dataflow/backend/router.py +++ b/src/govoplan_dataflow/backend/router.py @@ -5,6 +5,10 @@ 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.dataflows import ( + DataflowPublicationTarget, + DataflowRunRequest, +) from govoplan_core.core.datasources import ( DatasourceAccessError, DatasourceDescriptor, @@ -32,6 +36,9 @@ from govoplan_dataflow.backend.schemas import ( PipelineListResponse, PipelinePreviewRequest, PipelinePreviewResponse, + PipelineRunCreateRequest, + PipelineRunListResponse, + PipelineRunResponse, PipelineResponse, PipelineSqlResponse, PipelineUpdateRequest, @@ -48,12 +55,17 @@ from govoplan_dataflow.backend.service import ( DataflowValidationError, compile_sql_draft, create_pipeline, + cancel_pipeline_run, delete_pipeline, get_pipeline, + get_pipeline_run, + list_pipeline_runs, list_pipelines, pipeline_response, + pipeline_run_response, preview_pipeline, render_graph_sql, + start_pipeline_run, update_pipeline, validate_draft, ) @@ -167,6 +179,7 @@ def _source_response(source: DatasourceDescriptor) -> TabularSourceResponse: source_name=source.source_name, name=source.name, description=source.description, + mode=source.mode, columns=[ TabularSourceColumnResponse( name=column.name, @@ -421,6 +434,152 @@ def api_delete_pipeline( return PipelineDeleteResponse(deleted=True, pipeline_id=pipeline.id) +@router.get( + "/pipelines/{pipeline_id}/runs", + response_model=PipelineRunListResponse, +) +def api_list_pipeline_runs( + pipeline_id: str, + limit: int = 50, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PipelineRunListResponse: + _require_any_scope(principal, READ_SCOPE, RUN_SCOPE, ADMIN_SCOPE) + try: + runs = list_pipeline_runs( + session, + tenant_id=principal.tenant_id, + pipeline_id=pipeline_id, + limit=limit, + ) + return PipelineRunListResponse( + runs=[pipeline_run_response(session, run) for run in runs] + ) + except DataflowError as exc: + raise _http_error(exc) from exc + + +@router.post( + "/pipelines/{pipeline_id}/runs", + response_model=PipelineRunResponse, + status_code=status.HTTP_201_CREATED, +) +def api_start_pipeline_run( + pipeline_id: str, + payload: PipelineRunCreateRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PipelineRunResponse: + _require_any_scope(principal, RUN_SCOPE, ADMIN_SCOPE) + try: + pipeline = get_pipeline( + session, + tenant_id=principal.tenant_id, + pipeline_id=pipeline_id, + ) + publication = ( + DataflowPublicationTarget( + target_datasource_ref=payload.publication.target_datasource_ref, + name=payload.publication.name, + source_name=payload.publication.source_name, + description=payload.publication.description, + freeze=payload.publication.freeze, + frozen_label=payload.publication.frozen_label, + set_current=payload.publication.set_current, + metadata=payload.publication.metadata, + ) + if payload.publication + else None + ) + run, replayed = start_pipeline_run( + session, + tenant_id=principal.tenant_id, + actor_id=_actor_id(principal), + principal=principal, + registry=get_registry(), + request=DataflowRunRequest( + pipeline_ref=f"pipeline:{pipeline.id}", + revision=payload.revision or pipeline.current_revision, + idempotency_key=payload.idempotency_key, + row_limit=payload.row_limit, + publication=publication, + ), + ) + 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.run.replayed" + if replayed + else f"dataflow.run.{run.status}" + ), + object_type="dataflow_run", + object_id=run.id, + details={ + "pipeline_id": run.pipeline_id, + "revision_id": run.pipeline_revision_id, + "status": run.status, + "output_datasource_ref": run.output_datasource_ref, + "output_materialization_ref": run.output_materialization_ref, + }, + ) + response = pipeline_run_response(session, run, replayed=replayed) + session.commit() + return response + + +@router.get("/runs/{run_id}", response_model=PipelineRunResponse) +def api_get_pipeline_run( + run_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PipelineRunResponse: + _require_any_scope(principal, READ_SCOPE, RUN_SCOPE, ADMIN_SCOPE) + try: + run = get_pipeline_run( + session, + tenant_id=principal.tenant_id, + run_ref=f"dataflow-run:{run_id}", + ) + return pipeline_run_response(session, run) + except DataflowError as exc: + raise _http_error(exc) from exc + + +@router.post("/runs/{run_id}/cancel", response_model=PipelineRunResponse) +def api_cancel_pipeline_run( + run_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PipelineRunResponse: + _require_any_scope(principal, RUN_SCOPE, ADMIN_SCOPE) + try: + run = cancel_pipeline_run( + session, + tenant_id=principal.tenant_id, + run_ref=f"dataflow-run:{run_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.run.cancelled", + object_type="dataflow_run", + object_id=run.id, + details={"pipeline_id": run.pipeline_id}, + ) + response = pipeline_run_response(session, run) + session.commit() + return response + + @router.post("/validate", response_model=PipelineValidationResponse) def api_validate_pipeline( payload: PipelineDraftRequest, diff --git a/src/govoplan_dataflow/backend/schemas.py b/src/govoplan_dataflow/backend/schemas.py index bbcbaf7..03fc678 100644 --- a/src/govoplan_dataflow/backend/schemas.py +++ b/src/govoplan_dataflow/backend/schemas.py @@ -158,6 +158,68 @@ class PipelinePreviewResponse(BaseModel): executor_version: str +class PipelineRunPublicationRequest(BaseModel): + target_datasource_ref: str | None = Field(default=None, max_length=500) + name: str | None = Field(default=None, min_length=1, max_length=300) + source_name: str | None = Field( + default=None, + min_length=1, + max_length=120, + pattern=r"^[A-Za-z_][A-Za-z0-9_]*$", + ) + description: str | None = Field(default=None, max_length=4_000) + freeze: bool = False + frozen_label: str | None = Field(default=None, max_length=300) + set_current: bool = True + metadata: dict[str, Any] = Field(default_factory=dict) + + @model_validator(mode="after") + def validate_target(self) -> "PipelineRunPublicationRequest": + if self.target_datasource_ref: + return self + if not self.name or not self.source_name: + raise ValueError( + "New publication targets require name and source_name." + ) + return self + + +class PipelineRunCreateRequest(BaseModel): + revision: int | None = Field(default=None, ge=1) + idempotency_key: str = Field(min_length=1, max_length=255) + row_limit: int = Field(default=500, ge=1, le=500) + publication: PipelineRunPublicationRequest | None = None + + +class PipelineRunResponse(BaseModel): + ref: str + pipeline_id: str + revision: int + run_type: str + status: Literal["running", "succeeded", "failed", "cancelled"] + idempotency_key: str | None + definition_hash: str + executor_version: str + source_fingerprints: list[dict[str, Any]] + result_schema: list[PreviewColumn] + diagnostics: list[DataflowDiagnostic] + input_row_count: int + output_row_count: int + output_publication_ref: str | None + output_datasource_ref: str | None + output_materialization_ref: str | None + error: str | None + started_at: datetime + finished_at: datetime | None + created_by: str | None + created_at: datetime + replayed: bool = False + + +class PipelineRunListResponse(BaseModel): + runs: list[PipelineRunResponse] + + class PipelineDeleteResponse(BaseModel): deleted: bool pipeline_id: str @@ -210,6 +272,7 @@ class TabularSourceResponse(BaseModel): source_name: str name: str description: str | None + mode: Literal["live", "cached", "static"] columns: list[TabularSourceColumnResponse] schema_version: str fingerprint: str diff --git a/src/govoplan_dataflow/backend/service.py b/src/govoplan_dataflow/backend/service.py index 5d43b2b..ad6dd89 100644 --- a/src/govoplan_dataflow/backend/service.py +++ b/src/govoplan_dataflow/backend/service.py @@ -1,15 +1,26 @@ from __future__ import annotations +import hashlib +import json from dataclasses import dataclass from sqlalchemy import select from sqlalchemy.orm import Session from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.dataflows import ( + DataflowRunConflictError, + DataflowRunDescriptor, + DataflowRunError, + DataflowRunNotFoundError, + DataflowRunRequest, +) from govoplan_core.core.datasources import ( DatasourceError, + DatasourcePublicationRequest, DatasourceReadRequest, datasource_catalogue, + datasource_publication, ) from govoplan_core.db.base import utcnow from govoplan_dataflow.backend.db.models import ( @@ -34,6 +45,7 @@ from govoplan_dataflow.backend.schemas import ( PipelinePreviewResponse, PipelineResponse, PipelineRevisionResponse, + PipelineRunResponse, PipelineSqlResponse, PipelineUpdateRequest, PipelineValidationResponse, @@ -45,15 +57,15 @@ from govoplan_dataflow.backend.sql_compiler import ( ) -class DataflowError(RuntimeError): +class DataflowError(DataflowRunError): pass -class DataflowNotFoundError(DataflowError): +class DataflowNotFoundError(DataflowError, DataflowRunNotFoundError): pass -class DataflowConflictError(DataflowError): +class DataflowConflictError(DataflowError, DataflowRunConflictError): pass @@ -495,6 +507,363 @@ def preview_pipeline( ) +def list_pipeline_runs( + session: Session, + *, + tenant_id: str, + pipeline_id: str | None = None, + limit: int = 100, +) -> list[DataflowRun]: + statement = ( + select(DataflowRun) + .where(DataflowRun.tenant_id == tenant_id) + .order_by(DataflowRun.created_at.desc(), DataflowRun.id.desc()) + .limit(max(1, min(int(limit), 100))) + ) + if pipeline_id: + get_pipeline( + session, + tenant_id=tenant_id, + pipeline_id=pipeline_id, + ) + statement = statement.where(DataflowRun.pipeline_id == pipeline_id) + return list(session.scalars(statement)) + + +def get_pipeline_run( + session: Session, + *, + tenant_id: str, + run_ref: str, +) -> DataflowRun: + run_id = _strip_ref(run_ref, "dataflow-run:") + if not run_id: + raise DataflowNotFoundError("Dataflow run not found") + run = session.scalar( + select(DataflowRun).where( + DataflowRun.id == run_id, + DataflowRun.tenant_id == tenant_id, + ) + ) + if run is None: + raise DataflowNotFoundError("Dataflow run not found") + return run + + +def start_pipeline_run( + session: Session, + *, + tenant_id: str, + actor_id: str | None, + principal: ApiPrincipal, + registry: object | None, + request: DataflowRunRequest, +) -> tuple[DataflowRun, bool]: + pipeline_id = _strip_ref(request.pipeline_ref, "pipeline:") + if not pipeline_id: + raise DataflowNotFoundError("Dataflow pipeline not found") + pipeline = get_pipeline( + session, + tenant_id=tenant_id, + pipeline_id=pipeline_id, + ) + revision = get_pipeline_revision( + session, + pipeline=pipeline, + revision=request.revision, + ) + idempotency_key = request.idempotency_key.strip() + if not idempotency_key or len(idempotency_key) > 255: + raise DataflowConflictError( + "A Dataflow run idempotency key of at most 255 characters is required." + ) + if request.row_limit < 1 or request.row_limit > 500: + raise DataflowConflictError( + "The bounded Dataflow runner supports between 1 and 500 output rows." + ) + request_hash = _run_request_hash(request) + existing = session.scalar( + select(DataflowRun).where( + DataflowRun.tenant_id == tenant_id, + DataflowRun.pipeline_id == pipeline.id, + DataflowRun.idempotency_key == idempotency_key, + ) + ) + if existing is not None: + if existing.request_hash != request_hash: + raise DataflowConflictError( + "The Dataflow run idempotency key was already used with " + "different parameters." + ) + return existing, True + + graph = PipelineGraph.model_validate(revision.graph) + started_at = utcnow() + run = DataflowRun( + tenant_id=tenant_id, + pipeline_id=pipeline.id, + pipeline_revision_id=revision.id, + run_type="published" if request.publication else "run", + status="running", + executor_version=EXECUTOR_VERSION, + definition_hash=revision.content_hash, + idempotency_key=idempotency_key, + request_hash=request_hash, + request_=_run_request_payload(request), + source_fingerprints=[], + result_schema=[], + diagnostics=[], + input_row_count=0, + output_row_count=0, + started_at=started_at, + created_by=actor_id, + ) + session.add(run) + session.flush() + + try: + result = execute_preview( + graph, + row_limit=request.row_limit, + source_resolver=_datasource_source_resolver( + session=session, + principal=principal, + registry=registry, + ), + ) + if request.publication and ( + result.truncated + or any( + bool(item.get("truncated")) + for item in result.source_fingerprints + ) + ): + raise PipelineExecutionError( + "The bounded runner cannot publish a truncated result or a " + "result calculated from truncated source data." + ) + + run.source_fingerprints = result.source_fingerprints + run.result_schema = [ + item.model_dump(mode="json") for item in result.columns + ] + run.diagnostics = [ + item.model_dump(mode="json") for item in result.diagnostics + ] + run.input_row_count = result.input_row_count + run.output_row_count = result.total_rows + + if request.publication: + publisher = datasource_publication(registry) + if publisher is None: + raise PipelineExecutionError( + "Publishing Dataflow output requires the Datasources " + "publication capability." + ) + target = request.publication + publication = publisher.publish_rows( + session, + principal, + request=DatasourcePublicationRequest( + producer_module="dataflow", + producer_run_ref=f"dataflow-run:{run.id}", + idempotency_key=f"{pipeline.id}:{idempotency_key}", + rows=tuple(dict(row) for row in result.rows), + target_datasource_ref=target.target_datasource_ref, + name=target.name or f"{pipeline.name} output", + source_name=target.source_name, + description=target.description, + freeze=target.freeze, + frozen_label=target.frozen_label, + set_current=target.set_current, + provenance={ + "pipeline_ref": f"pipeline:{pipeline.id}", + "pipeline_revision": revision.revision, + "definition_hash": revision.content_hash, + "source_fingerprints": result.source_fingerprints, + }, + metadata={ + **dict(target.metadata), + "dataflow_run_ref": f"dataflow-run:{run.id}", + }, + ), + ) + run.output_publication_ref = publication.ref + run.output_datasource_ref = publication.datasource.ref + run.output_materialization_ref = publication.materialization.ref + run.status = "succeeded" + run.finished_at = utcnow() + run.error = None + except (DatasourceError, PipelineExecutionError) as exc: + run.status = "failed" + run.finished_at = utcnow() + run.error = str(exc) + diagnostics = list(getattr(exc, "diagnostics", ())) + diagnostics.append( + DataflowDiagnostic( + severity="error", + code="run.execution", + message=str(exc), + node_id=getattr(exc, "node_id", None), + ) + ) + run.diagnostics = [ + item.model_dump(mode="json") for item in diagnostics + ] + run.source_fingerprints = list( + getattr(exc, "source_fingerprints", ()) + ) + run.input_row_count = int(getattr(exc, "input_row_count", 0)) + session.flush() + return run, False + + +def cancel_pipeline_run( + session: Session, + *, + tenant_id: str, + run_ref: str, +) -> DataflowRun: + run = get_pipeline_run( + session, + tenant_id=tenant_id, + run_ref=run_ref, + ) + if run.status not in {"queued", "running"}: + raise DataflowConflictError( + f"Dataflow run is already {run.status} and cannot be cancelled." + ) + run.status = "cancelled" + run.finished_at = utcnow() + run.error = "Cancelled by request." + session.flush() + return run + + +def pipeline_run_response( + session: Session, + run: DataflowRun, + *, + replayed: bool = False, +) -> PipelineRunResponse: + revision = session.get(DataflowPipelineRevision, run.pipeline_revision_id) + if revision is None: + raise DataflowNotFoundError("Dataflow pipeline revision not found") + return PipelineRunResponse( + ref=f"dataflow-run:{run.id}", + pipeline_id=run.pipeline_id, + revision=revision.revision, + run_type=run.run_type, + status=run.status, # type: ignore[arg-type] + idempotency_key=run.idempotency_key, + definition_hash=run.definition_hash, + executor_version=run.executor_version, + source_fingerprints=list(run.source_fingerprints), + result_schema=list(run.result_schema), + diagnostics=list(run.diagnostics), + input_row_count=run.input_row_count, + output_row_count=run.output_row_count, + output_publication_ref=run.output_publication_ref, + output_datasource_ref=run.output_datasource_ref, + output_materialization_ref=run.output_materialization_ref, + error=run.error, + started_at=run.started_at, + finished_at=run.finished_at, + created_by=run.created_by, + created_at=run.created_at, + replayed=replayed, + ) + + +def pipeline_run_descriptor( + session: Session, + run: DataflowRun, + *, + replayed: bool = False, +) -> DataflowRunDescriptor: + revision = session.get(DataflowPipelineRevision, run.pipeline_revision_id) + if revision is None: + raise DataflowNotFoundError("Dataflow pipeline revision not found") + return DataflowRunDescriptor( + ref=f"dataflow-run:{run.id}", + pipeline_ref=f"pipeline:{run.pipeline_id}", + revision=revision.revision, + status=run.status, + definition_hash=run.definition_hash, + executor_version=run.executor_version, + input_row_count=run.input_row_count, + output_row_count=run.output_row_count, + output_publication_ref=run.output_publication_ref, + output_datasource_ref=run.output_datasource_ref, + output_materialization_ref=run.output_materialization_ref, + error=run.error, + started_at=run.started_at, + finished_at=run.finished_at, + replayed=replayed, + metadata={ + "run_type": run.run_type, + "source_fingerprints": list(run.source_fingerprints), + "diagnostics": list(run.diagnostics), + }, + ) + + +class SqlDataflowRunLifecycleProvider: + def __init__(self, *, registry: object | None = None) -> None: + self._registry = registry + + def start_run( + self, + session: object, + principal: object, + *, + request: DataflowRunRequest, + ) -> DataflowRunDescriptor: + db, api_principal = _run_context(session, principal) + run, replayed = start_pipeline_run( + db, + tenant_id=api_principal.tenant_id, + actor_id=_principal_actor_id(api_principal), + principal=api_principal, + registry=self._registry, + request=request, + ) + return pipeline_run_descriptor(db, run, replayed=replayed) + + def get_run( + self, + session: object, + principal: object, + *, + run_ref: str, + ) -> DataflowRunDescriptor | None: + db, api_principal = _run_context(session, principal) + try: + run = get_pipeline_run( + db, + tenant_id=api_principal.tenant_id, + run_ref=run_ref, + ) + except DataflowNotFoundError: + return None + return pipeline_run_descriptor(db, run) + + def cancel_run( + self, + session: object, + principal: object, + *, + run_ref: str, + ) -> DataflowRunDescriptor: + db, api_principal = _run_context(session, principal) + run = cancel_pipeline_run( + db, + tenant_id=api_principal.tenant_id, + run_ref=run_ref, + ) + return pipeline_run_descriptor(db, run) + + def normalize_definition( *, graph: PipelineGraph, @@ -541,6 +910,109 @@ def _source_nodes( return nodes +def _datasource_source_resolver( + *, + session: Session, + principal: ApiPrincipal, + registry: object | None, +): + provider = datasource_catalogue(registry) + + def resolve_source(node: GraphNode, limit: int) -> ResolvedSource: + if provider is None: + raise PipelineExecutionError( + "Datasource-backed execution requires the Datasources " + "catalogue capability.", + node_id=node.id, + ) + try: + resolved = provider.read_datasource( + session, + principal, + request=DatasourceReadRequest( + datasource_ref=str(node.config["source_ref"]), + consistency=str( + node.config.get("consistency") or "current" + ), # type: ignore[arg-type] + limit=limit, + expected_fingerprint=_clean_optional( + node.config.get("expected_fingerprint") + ), + ), + ) + except DatasourceError as exc: + raise PipelineExecutionError(str(exc), node_id=node.id) from exc + return ResolvedSource( + rows=tuple(dict(row) for row in resolved.rows), + source_ref=resolved.datasource.ref, + provider=resolved.datasource.provider or "datasources", + fingerprint=resolved.datasource.fingerprint, + total_rows=resolved.total_rows, + truncated=resolved.truncated, + ) + + return resolve_source + + +def _run_context( + session: object, + principal: object, +) -> tuple[Session, ApiPrincipal]: + if not isinstance(session, Session): + raise TypeError("Dataflow run providers require a SQLAlchemy session.") + if not isinstance(principal, ApiPrincipal): + raise DataflowConflictError("A tenant API principal is required.") + if not principal.tenant_id: + raise DataflowConflictError("A tenant API principal is required.") + return session, principal + + +def _principal_actor_id(principal: ApiPrincipal) -> str | None: + return principal.account_id or principal.membership_id or principal.identity_id + + +def _strip_ref(value: str, prefix: str) -> str | None: + cleaned = str(value or "").strip() + if not cleaned: + return None + if cleaned.startswith(prefix): + return cleaned[len(prefix) :] + return cleaned if ":" not in cleaned else None + + +def _run_request_payload(request: DataflowRunRequest) -> dict[str, object]: + target = request.publication + return { + "pipeline_ref": request.pipeline_ref, + "revision": request.revision, + "row_limit": request.row_limit, + "publication": ( + { + "target_datasource_ref": target.target_datasource_ref, + "name": target.name, + "source_name": target.source_name, + "description": target.description, + "freeze": target.freeze, + "frozen_label": target.frozen_label, + "set_current": target.set_current, + "metadata": dict(target.metadata), + } + if target + else None + ), + } + + +def _run_request_hash(request: DataflowRunRequest) -> str: + encoded = json.dumps( + _run_request_payload(request), + sort_keys=True, + separators=(",", ":"), + default=str, + ) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + def _clean_optional(value: str | None) -> str | None: if value is None: return None @@ -553,16 +1025,23 @@ __all__ = [ "DataflowError", "DataflowNotFoundError", "DataflowValidationError", + "SqlDataflowRunLifecycleProvider", + "cancel_pipeline_run", "compile_sql_draft", "create_pipeline", "delete_pipeline", "get_pipeline", "get_pipeline_revision", + "get_pipeline_run", + "list_pipeline_runs", "list_pipelines", "normalize_definition", "pipeline_response", + "pipeline_run_descriptor", + "pipeline_run_response", "preview_pipeline", "render_graph_sql", + "start_pipeline_run", "update_pipeline", "validate_draft", ] diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 907a619..13c50ba 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -2,6 +2,7 @@ from __future__ import annotations import unittest +from govoplan_core.core.dataflows import CAPABILITY_DATAFLOW_RUN_LIFECYCLE from govoplan_core.core.modules import ModuleManifest from govoplan_dataflow.backend.manifest import get_manifest @@ -19,6 +20,10 @@ class DataflowManifestTests(unittest.TestCase): self.assertEqual(manifest.frontend.package_name, "@govoplan/dataflow-webui") self.assertIsNotNone(manifest.route_factory) self.assertIsNotNone(manifest.migration_spec) + self.assertIn( + CAPABILITY_DATAFLOW_RUN_LIFECYCLE, + manifest.capability_factories, + ) self.assertEqual( { "dataflow.pipeline_catalog", diff --git a/tests/test_migrations.py b/tests/test_migrations.py index d46a8b2..37bc208 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( - "d4f7a1c8e2b0", + "b8e3c6a1f4d9", set(MigrationContext.configure(connection).get_current_heads()), ) self.assertEqual( diff --git a/tests/test_service.py b/tests/test_service.py index 2466bad..3440691 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -5,6 +5,18 @@ import unittest from sqlalchemy import create_engine, func, select from sqlalchemy.orm import Session, sessionmaker +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.dataflows import ( + DataflowPublicationTarget, + DataflowRunRequest, +) +from govoplan_core.core.datasources import ( + CAPABILITY_DATASOURCE_PUBLICATION, + DatasourceDescriptor, + DatasourceMaterialization, + DatasourcePublicationResult, +) from govoplan_core.db.base import Base from govoplan_dataflow.backend.db.models import ( DataflowPipeline, @@ -28,6 +40,7 @@ from govoplan_dataflow.backend.service import ( get_pipeline, list_pipelines, preview_pipeline, + start_pipeline_run, update_pipeline, ) @@ -71,6 +84,61 @@ def sample_graph(*, minimum: int = 10) -> PipelineGraph: ) +def principal(tenant_id: str = "tenant-1") -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id=tenant_id, + scopes=frozenset(), + ), + account=object(), + user=object(), + ) + + +class FakePublicationProvider: + def __init__(self) -> None: + self.requests = [] + + def publish_rows(self, _session, _principal, *, request): + self.requests.append(request) + descriptor = DatasourceDescriptor( + ref=request.target_datasource_ref or "datasource:output-1", + source_name=request.source_name or "existing_output", + name=request.name or "Existing output", + kind="custom", + mode="static", + shape="tabular", + fingerprint="published-fingerprint", + ) + return DatasourcePublicationResult( + ref="publication:publication-1", + status="published", + datasource=descriptor, + materialization=DatasourceMaterialization( + ref="materialization:materialization-1", + datasource_ref=descriptor.ref, + revision=1, + state="published", + fingerprint=descriptor.fingerprint, + ), + ) + + +class FakeRegistry: + def __init__(self, publication_provider: FakePublicationProvider) -> None: + self.publication_provider = publication_provider + + def has_capability(self, name: str) -> bool: + return name == CAPABILITY_DATASOURCE_PUBLICATION + + def capability(self, name: str): + if not self.has_capability(name): + raise KeyError(name) + return self.publication_provider + + class DataflowServiceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:") @@ -263,6 +331,109 @@ class DataflowServiceTests(unittest.TestCase): self.assertEqual(run.source_fingerprints, response.source_fingerprints) self.assertEqual("preview.execution", run.diagnostics[-1]["code"]) + def test_pinned_run_publishes_once_and_replays_idempotently(self) -> None: + pipeline = self._create() + publication_provider = FakePublicationProvider() + request = DataflowRunRequest( + pipeline_ref=f"pipeline:{pipeline.id}", + revision=1, + idempotency_key="monthly-2026-07", + publication=DataflowPublicationTarget( + name="Monthly result", + source_name="monthly_result", + freeze=True, + frozen_label="July 2026", + ), + ) + + first, first_replayed = start_pipeline_run( + self.session, + tenant_id="tenant-1", + actor_id="user-1", + principal=principal(), + registry=FakeRegistry(publication_provider), + request=request, + ) + replay, replayed = start_pipeline_run( + self.session, + tenant_id="tenant-1", + actor_id="user-1", + principal=principal(), + registry=FakeRegistry(publication_provider), + request=request, + ) + self.session.commit() + + self.assertFalse(first_replayed) + self.assertTrue(replayed) + self.assertEqual(first.id, replay.id) + self.assertEqual("succeeded", first.status) + self.assertEqual("datasource:output-1", first.output_datasource_ref) + self.assertEqual( + "materialization:materialization-1", + first.output_materialization_ref, + ) + self.assertEqual( + [{"id": 2, "amount": 15}, {"id": 3, "amount": 25}], + list(publication_provider.requests[0].rows), + ) + self.assertEqual(1, len(publication_provider.requests)) + + def test_run_idempotency_key_rejects_changed_parameters(self) -> None: + pipeline = self._create() + publication_provider = FakePublicationProvider() + registry = FakeRegistry(publication_provider) + start_pipeline_run( + self.session, + tenant_id="tenant-1", + actor_id="user-1", + principal=principal(), + registry=registry, + request=DataflowRunRequest( + pipeline_ref=f"pipeline:{pipeline.id}", + revision=1, + idempotency_key="stable-key", + row_limit=100, + ), + ) + + with self.assertRaises(DataflowConflictError): + start_pipeline_run( + self.session, + tenant_id="tenant-1", + actor_id="user-1", + principal=principal(), + registry=registry, + request=DataflowRunRequest( + pipeline_ref=f"pipeline:{pipeline.id}", + revision=1, + idempotency_key="stable-key", + row_limit=10, + ), + ) + + def test_publication_without_datasources_finishes_as_failed_run(self) -> None: + pipeline = self._create() + run, _ = start_pipeline_run( + self.session, + tenant_id="tenant-1", + actor_id="user-1", + principal=principal(), + registry=None, + request=DataflowRunRequest( + pipeline_ref=f"pipeline:{pipeline.id}", + revision=1, + idempotency_key="missing-publisher", + publication=DataflowPublicationTarget( + name="Output", + source_name="output", + ), + ), + ) + + self.assertEqual("failed", run.status) + self.assertIn("Datasources publication capability", run.error) + if __name__ == "__main__": unittest.main() diff --git a/webui/src/api/dataflow.ts b/webui/src/api/dataflow.ts index 86a47b2..55cf9e7 100644 --- a/webui/src/api/dataflow.ts +++ b/webui/src/api/dataflow.ts @@ -146,6 +146,7 @@ export type TabularSource = { source_name: string; name: string; description?: string | null; + mode: "live" | "cached" | "static"; columns: TabularSourceColumn[]; schema_version: string; fingerprint: string; @@ -161,6 +162,31 @@ export type TabularSourceCatalogue = { sources: TabularSource[]; }; +export type PipelineRun = { + ref: string; + pipeline_id: string; + revision: number; + run_type: string; + status: "running" | "succeeded" | "failed" | "cancelled"; + idempotency_key?: string | null; + definition_hash: string; + executor_version: string; + source_fingerprints: Record[]; + result_schema: PreviewColumn[]; + diagnostics: DataflowDiagnostic[]; + input_row_count: number; + output_row_count: number; + output_publication_ref?: string | null; + output_datasource_ref?: string | null; + output_materialization_ref?: string | null; + error?: string | null; + started_at: string; + finished_at?: string | null; + created_by?: string | null; + created_at: string; + replayed: boolean; +}; + export async function listDataflowNodeTypes(settings: ApiSettings): Promise { const response = await apiFetch<{ nodes: NodeTypeDefinition[] }>( settings, @@ -280,3 +306,43 @@ export function previewDataflowPipeline( body: JSON.stringify(payload) }); } + +export async function listDataflowPipelineRuns( + settings: ApiSettings, + pipelineId: string +): Promise { + const response = await apiFetch<{ runs: PipelineRun[] }>( + settings, + `/api/v1/dataflow/pipelines/${encodeURIComponent(pipelineId)}/runs?limit=25` + ); + return response.runs; +} + +export function runDataflowPipeline( + settings: ApiSettings, + pipelineId: string, + payload: { + revision: number; + idempotency_key: string; + row_limit?: number; + publication?: { + target_datasource_ref?: string | null; + name?: string | null; + source_name?: string | null; + description?: string | null; + freeze?: boolean; + frozen_label?: string | null; + set_current?: boolean; + metadata?: Record; + } | null; + } +): Promise { + return apiFetch( + settings, + `/api/v1/dataflow/pipelines/${encodeURIComponent(pipelineId)}/runs`, + { + method: "POST", + body: JSON.stringify(payload) + } + ); +} diff --git a/webui/src/features/dataflow/DataflowPage.tsx b/webui/src/features/dataflow/DataflowPage.tsx index 5d81ec1..431ae5e 100644 --- a/webui/src/features/dataflow/DataflowPage.tsx +++ b/webui/src/features/dataflow/DataflowPage.tsx @@ -8,6 +8,7 @@ import { import { CheckCircle2, Code2, + DatabaseZap, Network, Play, Plus, @@ -28,6 +29,7 @@ import { LoadingFrame, SegmentedControl, StatusBadge, + ToggleSwitch, hasScope, isApiError, useUnsavedChanges, @@ -42,9 +44,11 @@ import { createDataflowSourceSnapshot, deleteDataflowPipeline, listDataflowNodeTypes, + listDataflowPipelineRuns, listDataflowPipelines, listDataflowSources, previewDataflowPipeline, + runDataflowPipeline, renderDataflowSql, updateDataflowPipeline, validateDataflowPipeline, @@ -56,6 +60,7 @@ import { type Pipeline, type PipelineGraphNode, type PipelinePreview, + type PipelineRun, type TabularSource } from "../../api/dataflow"; import DataflowCanvas, { updateGraphNode } from "./DataflowCanvas"; @@ -96,6 +101,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings const [resultTab, setResultTab] = useState("preview"); const [deleteOpen, setDeleteOpen] = useState(false); const [snapshotOpen, setSnapshotOpen] = useState(false); + const [runOpen, setRunOpen] = useState(false); const [nodeLibrary, setNodeLibrary] = useState(FALLBACK_NODE_LIBRARY); const [sources, setSources] = useState([]); const [sourceCatalogueAvailable, setSourceCatalogueAvailable] = useState(false); @@ -533,6 +539,13 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings + } @@ -724,10 +737,265 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings setSelectedNodeId(node.id); }} /> + setRunOpen(false)} + onCompleted={(run) => { + if (run.status === "succeeded") { + setSuccess( + run.output_datasource_ref + ? `Run published ${run.output_row_count} rows to ${run.output_datasource_ref}.` + : `Run completed with ${run.output_row_count} output rows.` + ); + } else { + setError(run.error || "Dataflow run failed."); + } + if (run.output_datasource_ref) { + void listDataflowSources(settings).then((catalogue) => { + setSources(catalogue.sources); + setSourceCatalogueAvailable(catalogue.available); + setSourceCatalogueWritable(catalogue.writable); + }); + } + }} + /> ); } +type RunMode = "run" | "publish"; + +function RunPipelineDialog({ + open, + settings, + pipeline, + sources, + onClose, + onCompleted +}: { + open: boolean; + settings: ApiSettings; + pipeline: { id: string; name: string; revision: number } | null; + sources: TabularSource[]; + onClose: () => void; + onCompleted: (run: PipelineRun) => void; +}) { + const [mode, setMode] = useState("run"); + const [targetRef, setTargetRef] = useState("new"); + const [name, setName] = useState(""); + const [sourceName, setSourceName] = useState(""); + const [description, setDescription] = useState(""); + const [freeze, setFreeze] = useState(false); + const [frozenLabel, setFrozenLabel] = useState(""); + const [idempotencyKey, setIdempotencyKey] = useState(""); + const [runs, setRuns] = useState([]); + const [loadingRuns, setLoadingRuns] = useState(false); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(""); + const publicationTargets = sources.filter((source) => + source.mode !== "live" && source.capabilities.includes("read") + ); + + useEffect(() => { + if (!open || !pipeline) return; + setMode("run"); + setTargetRef("new"); + setName(`${pipeline.name} output`); + setSourceName(sourceNameFromFile(`${pipeline.name}_output`)); + setDescription(""); + setFreeze(false); + setFrozenLabel(""); + setIdempotencyKey(crypto.randomUUID()); + setError(""); + setLoadingRuns(true); + void listDataflowPipelineRuns(settings, pipeline.id) + .then(setRuns) + .catch((loadError) => setError(apiErrorMessage(loadError))) + .finally(() => setLoadingRuns(false)); + }, [open, pipeline?.id, pipeline?.revision, settings]); + + const start = async () => { + if (!pipeline) return; + setBusy(true); + setError(""); + try { + const run = await runDataflowPipeline(settings, pipeline.id, { + revision: pipeline.revision, + idempotency_key: idempotencyKey, + row_limit: 500, + publication: mode === "publish" ? { + target_datasource_ref: targetRef === "new" ? null : targetRef, + name: targetRef === "new" ? name.trim() : null, + source_name: targetRef === "new" ? sourceName.trim() : null, + description: targetRef === "new" ? description.trim() || null : null, + freeze, + frozen_label: freeze ? frozenLabel.trim() || null : null, + set_current: true + } : null + }); + setRuns((current) => [ + run, + ...current.filter((item) => item.ref !== run.ref) + ]); + onCompleted(run); + setIdempotencyKey(crypto.randomUUID()); + if (run.status !== "succeeded") { + setError(run.error || "Dataflow run failed."); + } + } catch (runError) { + setError(apiErrorMessage(runError)); + } finally { + setBusy(false); + } + }; + + const targetComplete = mode === "run" + || targetRef !== "new" + || Boolean(name.trim() && sourceName.trim()); + + return ( + { + if (!busy) onClose(); + }} + footer={( + <> + + + + )} + > +
+ {error ? ( + + {error} + + ) : null} +
+ + ariaLabel="Run output" + options={[ + { id: "run", label: "Run only" }, + { id: "publish", label: "Publish result" } + ]} + value={mode} + onChange={setMode} + /> + + Revision {pipeline?.revision ?? "-"} + +
+ {mode === "publish" ? ( +
+ + + + {targetRef === "new" ? ( + <> + + setName(event.target.value)} + /> + + + setSourceName(event.target.value)} + pattern="[A-Za-z_][A-Za-z0-9_]*" + /> + + + setDescription(event.target.value)} + /> + + + ) : null} +
+ +
+ {freeze ? ( + + setFrozenLabel(event.target.value)} + /> + + ) : null} +
+ ) : null} +
+
+ Recent runs + {loadingRuns ? "Loading..." : `${runs.length} shown`} +
+
+ + + + + + + + + + + + {runs.map((run) => ( + + + + + + + + ))} + {!loadingRuns && !runs.length ? ( + + ) : null} + +
StartedRevisionStatusRowsOutput
{new Date(run.started_at).toLocaleString()}{run.revision}{run.output_row_count.toLocaleString()} + {run.output_datasource_ref ?? "Run evidence"} +
No runs yet
+
+
+
+ + ); +} + function SourceSnapshotDialog({ open, settings, diff --git a/webui/src/styles/dataflow.css b/webui/src/styles/dataflow.css index 4217f7b..4dee0aa 100644 --- a/webui/src/styles/dataflow.css +++ b/webui/src/styles/dataflow.css @@ -601,6 +601,106 @@ width: min(720px, calc(100vw - 32px)); } +.dataflow-run-dialog { + width: min(860px, calc(100vw - 32px)); +} + +.dataflow-run-dialog-content { + display: grid; + gap: 14px; + max-height: min(680px, calc(100vh - 210px)); + overflow: hidden; +} + +.dataflow-run-controls, +.dataflow-run-history-heading { + display: flex; + align-items: center; + justify-content: space-between; + gap: 10px; +} + +.dataflow-run-revision, +.dataflow-run-history-heading small { + color: var(--muted); + font-size: 11px; +} + +.dataflow-run-publication-fields { + display: grid; + grid-template-columns: repeat(2, minmax(0, 1fr)); + gap: 12px; + padding: 12px 0; + border-top: var(--border-line); + border-bottom: var(--border-line); +} + +.dataflow-run-publication-fields input, +.dataflow-run-publication-fields select { + margin-top: 5px; + padding: 7px 8px; +} + +.dataflow-run-freeze { + display: flex; + min-height: 54px; + align-items: end; + padding-bottom: 4px; +} + +.dataflow-run-history { + display: flex; + min-height: 190px; + flex-direction: column; + overflow: hidden; +} + +.dataflow-run-history-heading { + min-height: 36px; + flex: 0 0 auto; + border-bottom: var(--border-line); +} + +.dataflow-run-history-heading strong { + color: var(--text-strong); + font-size: 12px; +} + +.dataflow-run-history > div:last-child { + min-height: 0; + flex: 1 1 auto; + overflow: auto; +} + +.dataflow-run-history table { + width: 100%; + border-collapse: collapse; + font-size: 11px; +} + +.dataflow-run-history th, +.dataflow-run-history td { + max-width: 230px; + border-bottom: var(--border-line); + padding: 7px 9px; + overflow: hidden; + text-align: left; + text-overflow: ellipsis; + white-space: nowrap; +} + +.dataflow-run-history th { + position: sticky; + z-index: 1; + top: 0; + background: var(--panel-soft); + color: var(--text-strong); +} + +.dataflow-run-history tbody tr:hover { + background: var(--primary-soft); +} + .dataflow-source-dialog-fields { display: grid; grid-template-columns: 1fr 1fr; @@ -942,4 +1042,8 @@ .dataflow-results { flex-basis: 220px; } + + .dataflow-run-publication-fields { + grid-template-columns: 1fr; + } }