Hooks and Steps Reference

Hooks are custom actions that execute at specific points in replications. Steps are the same concept used in pipelines.

Hook Locations (Replications)

Configuration

Replication-Level Hooks

hooks:
  start:
    - type: log
      message: "Starting replication"
  end:
    - type: check
      check: execution.status.error == 0

defaults:
  ...
streams:
  ...

Stream-Level Hooks

streams:
  my_table:
    hooks:
      pre:
        - type: query
          connection: source
          query: "SELECT 1"
      post:
        - type: http
          url: "https://webhook.example.com"

Common Properties

Hook Types

log

- type: log
  level: info  # debug, info, warn, error
  message: "Processed {run.total_rows} rows"

query

- type: query
  connection: MY_DB
  query: |
    UPDATE status SET last_run = NOW()
    WHERE table_name = '{stream.table}'
  into: result  # Optional: store result
  id: update_status

http

- type: http
  url: "https://api.example.com/webhook"
  method: POST
  headers:
    Authorization: "Bearer {env.API_TOKEN}"
    Content-Type: "application/json"
  payload: |
    {
      "table": "{stream.table}",
      "rows": {run.total_rows},
      "status": "{run.status}"
    }
  into: response

check

- type: check
  check: "run.total_rows > 0"
  failure_message: "No rows processed"
  on_failure: abort
  vars:
    threshold: 100

command

- type: command
  command: "python validate.py {object.table}"
  working_dir: "/scripts"
  print: true
  into: output

copy

- type: copy
  from: "file://data/source.csv"
  to: "s3://bucket/archive/{YYYY}/{MM}/"
  recursive: true

delete

- type: delete
  location: "s3://bucket/temp/"
  recursive: true

write

- type: write
  to: "file://logs/run_{timestamp.file_name}.txt"
  content: |
    Run completed at {timestamp.datetime}
    Rows: {run.total_rows}
    Status: {run.status}

read

- type: read
  from: "file://config/settings.json"
  into: settings

list

- type: list
  location: "s3://bucket/data/"
  recursive: false
  only: files  # files or folders
  id: file_list

store

- type: store
  key: my_value
  value: "{run.total_rows}"

replication

- type: replication
  path: /path/to/other_replication.yaml
  select_streams: ["table1"]
  mode: incremental

group

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

Variables

Variables use {variable.path} syntax within hook properties.

Debug: Print All Variables

- type: log
  message: '{runtime_state}'  # Prints all available variables as JSON

Variable Scopes

Commonly Used Variables

# Timestamps
timestamp.date              # "2025-01-19"
timestamp.datetime          # "2025-01-19 08:27:31"
timestamp.YYYY / MM / DD    # Year, month, day parts
timestamp.file_name         # "2025_01_19_082731" (safe for filenames)
timestamp.unix              # 1737286051

# Run metrics
run.total_rows              # Rows processed
run.total_bytes             # Bytes processed
run.status                  # "success" or "error"
run.duration                # Seconds

# Execution totals
execution.total_rows        # All rows across streams
execution.status.error      # Count of failed streams
execution.status.success    # Count of successful streams

# Stream/Object names
stream.table / stream.schema      # Source table info
object.table / object.full_name   # Target table info

Error Handling

on_failure Options

Check Pattern

hooks:
  end:
    # Stop if any errors
    - type: check
      check: execution.status.error == 0
      on_failure: break

    # Only run if check passed
    - type: http
      url: "https://webhook.example.com"
      payload: '{"status": "success"}'

Complete Example

hooks:
  start:
    - type: log
      message: "Starting replication at {timestamp.datetime}"

    - type: query
      connection: '{source.name}'
      query: "SELECT MAX(updated_at) as max_date FROM audit"
      into: audit_check
      id: pre_check

  end:
    - type: check
      check: execution.status.error == 0
      on_failure: break

    - type: http
      url: "{env.SLACK_WEBHOOK}"
      method: POST
      payload: |
        {
          "text": "Replicated {execution.total_rows} rows"
        }

defaults:
  hooks:
    pre:
      - type: log
        message: "Starting {stream.table}"

    post:
      - type: log
        message: "{stream.table}: {run.total_rows} rows in {run.duration}s"

streams:
  public.users:
    hooks:
      post:
        - type: query
          connection: '{target.name}'
          query: "ANALYZE {object.full_name}"

Full Documentation

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