Pipelines

Pipelines are multi-step workflows that execute a sequence of tasks with control flow.

Pipeline steps and replication hooks are the same engine — see STEPS.md for the full reference of every step/hook type, all variable scopes, and error handling. This file covers pipeline authoring.

Driving pipelines from Python? Load the sling-python skill — the Pipeline class accepts typed Step* objects or the same dicts as the YAML.

Gather first

Resolve each row before you write YAML, in this order: the user's request, existing files, discovery (MCP / CLI), then the default in the table. Ask the user only for rows that stay unresolved, one at a time, with a proposed default. If every row resolves, do not ask — build, and state the assumptions you took. Confirm first only for destructive resolutions (for example a mode that drops or rewrites existing tables).

When you update a pipeline, read the file first. Preserve id: values where possible.

When to Use Pipelines

• Run tasks before/after replications
• Run queries, commands and checks
• Chain multiple replications
• Implement data validation
• Send notifications
• Manage file operations

Basic Structure

env:
  MY_VAR: "value"

steps:
  - type: log
    message: "Starting pipeline"

  - type: replication
    path: /path/to/replication.yaml
    id: main_repl

  - type: query
    connection: MY_DB
    query: "SELECT COUNT(*) FROM table"
    if: state.main_repl.status == "success"

Validate, then run

Validate (MCP)

{"action": "validate", "input": {"file_path": "/path/to/pipeline.yaml"}}

Run (CLI only)

There is no MCP run action. Execute with the CLI:

sling run -p /path/to/pipeline.yaml
sling run -p /path/to/pipeline.yaml --debug

SQL models use sling build run (there is no MCP build action).

Step Types

See STEPS.md for full syntax and options of each type.

Pipeline spine

In a Sling project, the daily pipeline is replicate → build → check → log. Use real step types only. Do not use notify — that type is not active. See PROJECT.md in the sling skill.

steps:
  - type: log
    message: "Start daily run"

  - type: replication
    path: replications/stripe.yaml
    id: load_raw

  - type: build
    build: models
    prod: true
    id: transform

  - type: check
    check: state.transform.failed == 0
    failure_message: "Build reported failures"
    on_failure: abort

  - type: log
    message: "Done. build ok={state.transform.ok}"

Paths are root-relative. Run from the project root.

Common Steps

log

- type: log
  level: info  # debug, info, warn, error
  message: "Processing {env.MY_VAR}"

replication

- type: replication
  path: /path/to/replication.yaml
  streams: ["users", "orders"]
  mode: "incremental"
  id: my_repl

query

- type: query
  connection: MY_POSTGRES
  query: |
    SELECT COUNT(*) as cnt FROM users
    WHERE created_at > NOW() - INTERVAL '1 day'
  into: result
  id: count_query
  # rows land at state.<id>.result AND store.<into>

http

- type: http
  url: "https://api.example.com/webhook"
  method: POST
  headers:
    # {env.X} reads the process env; declare env: only to pin/override
    Authorization: "Bearer {env.API_TOKEN}"
  payload: |
    {"status": "{state.my_repl.status}"}
  into: response

command

- type: command
  command: "python validate.py {store.file_path}"
  working_dir: "/scripts"
  into: output

check

- type: check
  check: "state.count_query.result[0].cnt > 0"
  failure_message: "No records found"
  on_failure: abort

copy

- type: copy
  from: "local/data/today/file.csv"
  to: "aws_s3/archive/{timestamp.YYYY}/{timestamp.MM}/file.csv"

CONN/path: local/rel is cwd-relative, local//abs is absolute. After list, copy {loop.value.location} (already CONN/path); {loop.value.path} is also accepted as a local relative path.

set

- type: set
  key: my_value
  value: "something"
# Later: {store.my_value}

group (with loop)

- type: group
  loop: ["users", "products", "orders"]
  steps:
    - type: log
      message: "Processing: {loop.value}"
    - type: query
      connection: SNOWFLAKE
      query: "ANALYZE {loop.value}"

Control Flow

Conditional Execution

- type: replication
  path: main.yaml
  id: main_job

- type: http
  url: "https://slack.com/webhook"
  payload: '{"text": "Success!"}'
  if: state.main_job.status == "success"

- type: http
  url: "https://slack.com/webhook"
  payload: '{"text": "Failed!"}'
  if: state.main_job.status == "error"

Looping

- type: list
  location: "local/data/*.csv"
  id: files
  only: files

- type: group
  loop: state.files.result
  steps:
    - type: log
      message: "Processing: {loop.value.name}"
    - type: copy
      from: "{loop.value.location}"   # CONN/path — also accepts {loop.value.path}
      to: "aws_s3/eval/{loop.value.name}"
      single_file: true

Error Handling

on_failure options: abort (default), error, warn, quiet, skip, break, retry, defer — see STEPS.md for behavior details.

- type: delete
  location: "local/tmp/old/"
  on_failure: warn  # Don't fail if files don't exist

Variables

Expressions

Expressions use function calls, not methods. x.split('.') and x.upper() are invalid and fail at run time. Use the built-in functions: split, split_part, upper, and the rest of the function library (see the FUNCTIONS.md topic in the sling-api-specs skill).

A common loop pattern derives a table name from a file name. Use split_part for the stem. Quote the whole YAML value in single quotes, so the separator needs no escaping:

object: 'main.{split_part(loop.value.name, ".", 0)}'

Complete Example

env:
  SLACK_WEBHOOK: "${SLACK_WEBHOOK_URL}"

steps:
  - type: log
    message: "Starting data pipeline"

  - type: replication
    path: replications/main.yaml
    id: main_sync

  - type: check
    check: state.main_sync.status == "success"
    on_failure: break

  - type: query
    connection: TARGET_DB
    query: "CALL refresh_materialized_views()"

  - type: http
    url: "{env.SLACK_WEBHOOK}"
    method: POST
    payload: |
      {"text": "Pipeline completed: {state.main_sync.total_rows} rows"}

  - type: log
    message: "Pipeline finished"

Full Documentation

See https://docs.slingdata.io/concepts/pipeline.md for complete reference.