Data Quality Gate with Soda Checks

id: soda-quality-gate-pipeline
namespace: company.team
description: |
  Gate a pipeline on Soda check results instead of hoping the data is
  fine. io.kestra.plugin.soda.Scan runs SodaCL checks with allowFailure
  set, an If task branches on the hasFailures output, and bad data
  alerts Slack and fails the execution explicitly while good data logs
  a clean verdict and lets downstream work proceed.

triggers:
  - id: after_load_window
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run the gate every morning right after the nightly loads should
      have landed.
    cron: "0 7 * * *"
    disabled: true

inputs:
  - id: warehouse_path
    type: STRING
    defaults: /data/warehouse.duckdb
    description: Persistent path for the DuckDB warehouse file. Point at a volume
      that survives task runs so the scan sees the same file the load wrote.

tasks:
  - id: load_orders
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: Stand-in for the real load step. Creates and refreshes a small
      orders table so the gate has data to judge; replace with your dlt,
      Airbyte, or JDBC ingestion task.
    url: "jdbc:duckdb:{{ inputs.warehouse_path }}"
    sql: |
      CREATE TABLE IF NOT EXISTS orders (
          order_id INTEGER,
          customer_id VARCHAR,
          amount DOUBLE,
          created_at TIMESTAMP
      );
      DELETE FROM orders;
      INSERT INTO orders VALUES
          (1, 'cust-001', 42.50, now()),
          (2, 'cust-002', 17.90, now()),
          (3, 'cust-003', 250.00, now());
    fetchType: NONE

  - id: gate_scan
    type: io.kestra.plugin.soda.Scan
    description: Execute declarative SodaCL checks against the loaded orders table.
      allowFailure keeps a failed check from short-circuiting the flow, so the
      If task can decide what a failure means instead of the scan deciding for
      it.
    allowFailure: true
    configuration:
      data_source kestra:
        type: duckdb
        path: "{{ inputs.warehouse_path }}"
    checks:
      checks for orders:
        - row_count > 0
        - missing_count(customer_id) = 0
        - missing_count(amount) = 0
    requirements:
      - soda-core-duckdb

  - id: route_gate
    type: io.kestra.plugin.core.flow.If
    description: The gate itself. Failed checks alert Slack and fail the execution
      explicitly; passing checks log a clean verdict and let downstream tasks
      run.
    condition: "{{ outputs.gate_scan.result.hasFailures }}"
    then:
      - id: alert_blocked
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Tell the channel the gate closed before the execution fails, naming
          the flow and execution so the failing check is one click away.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Quality gate BLOCKED in flow {{ flow.id }}: Soda found failing checks on the orders table across {{ outputs.gate_scan.result.checks | length }} checks. Downstream tasks will not run (execution {{ execution.id }})."
          }
      - id: fail_execution
        type: io.kestra.plugin.core.execution.Fail
        description: Fail the execution so anything gated on this flow, such as
          transformation or publishing pipelines, stops instead of consuming bad
          data.
        errorMessage: "Soda quality gate failed on the orders table. Inspect the
          gate_scan task logs for the failing check before rerunning."
    else:
      - id: proceed_downstream
        type: io.kestra.plugin.core.log.Log
        description: Stand-in for the downstream work the gate protects; replace with
          your dbt run, export, or Subflow task.
        message: "Quality gate passed: {{ outputs.gate_scan.result.checks | length }}
          checks green on the orders table, downstream tasks may proceed."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the gate closes or the scan itself cannot run, so a
      blocked pipeline is loud instead of silently stuck.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Soda quality gate flow {{ flow.id }} FAILED (execution {{ execution.id }}). Either the gate blocked bad data or the scan could not run; inspect the gate_scan task logs."
      }