cannectors
ModulesInputs

database (input)

Pull records from PostgreSQL, MySQL, or SQLite.

The database input executes a SQL query, returns one record per row, and feeds them into the filter chain. Supports cursor or limit-offset pagination, incremental queries via a tracked timestamp/ID, and CRON scheduling.

Minimal example

input:
  type: database
  connectionStringRef: ${SOURCE_DATABASE_URL}
  query: SELECT id, name, email FROM customers

Options

propertytypedefaultdescription
typerequired
"database"—Module type discriminator. Must be `database` for this module.
id
string—Unique identifier within the pipeline.
name
string—Human-readable name.
description
string——
enabled
booleantrueWhether module is active.
tags
array<string>——
onError
string"fail"Default error action. Case-insensitive; normalized to lowercase by the runtime.
schedule
string—Optional CRON expression to run the database input on a schedule. Validated at runtime.
pagination
object——
incremental
object——
query
string—SQL query as a Jinja template rendered to text (control blocks {% if %}/{% for %} allowed). Dynamic values are never interpolated: use $1, $2, ... placeholders bound to the 'parameters' list.
queryFile
string—Path to a SQL file containing the query template (same contract as 'query').
parameters
array<string>—Ordered expr expressions bound to the query's $N placeholders (parameters[0] -> $1). Evaluated against record/meta/state/pagination; values bind natively typed (nil -> NULL).
connectionString
string—Database connection string (DSN). Avoid in production - use connectionStringRef instead.
connectionStringRef
string—Environment variable reference for connection string. Format: ${ENV_VAR_NAME}
driver
string—Database driver. Auto-detected from connection string if not specified.
postgresmysqlsqlite
maxOpenConns
integer10Maximum number of open connections in the pool.
maxIdleConns
integer5Maximum number of idle connections in the pool.
connMaxLifetimeSeconds
integer1800Maximum lifetime of a connection in seconds.
connMaxIdleTimeSeconds
integer300Maximum idle time for a connection in seconds.
timeoutMs
integer30000Query timeout in milliseconds.

Connection

Provide one of connectionString or connectionStringRef. Use the ref form in production — it reads from the environment, keeping secrets out of the YAML.

connectionStringRef: ${SOURCE_DATABASE_URL}

The driver is auto-detected from the URL scheme:

SchemeDriver
postgres://…postgres
mysql://… or user:pass@tcp(host:port)/dbmysql
file:./db.sqlitesqlite

Explicitly set driver: if you need to override the detection.

Pagination

Page values are exposed to parameter expressions as pagination.offset, pagination.cursor, and pagination.limit, re-evaluated on every page.

With type: limit-offset, binding those values is optional: when no parameter references pagination, the runtime appends LIMIT/OFFSET to the query automatically.

With type: cursor at least one parameter must reference pagination — the cursor can only reach the query through a parameter expression, and without one every page would re-run the same query. Pipelines that omit it are rejected at startup.

Limit + offset

pagination:
  type: limit-offset
  limit: 500
query: |
  SELECT id, payload
  FROM events
  ORDER BY id
  LIMIT $1 OFFSET $2
parameters:
  - pagination.limit
  - pagination.offset

Cursor

pagination:
  type: cursor
  cursorField: id        # which column to use as the cursor value
  limit: 500
query: |
  SELECT id, payload
  FROM events
  WHERE id > $1
  ORDER BY id
  LIMIT $2
parameters:
  - pagination.cursor ?? 0   # nil on the first page — coalesce a starting bound
  - pagination.limit

Cursor pagination is the right choice for any large incremental dataset — offsets get slow once you're millions of rows in.

Incremental queries

For queries that should only return rows changed since the last run, enable incremental and consume the persisted state through parameter expressions — state.lastRunTimestamp (RFC3339) and state.lastRunId:

incremental:
  enabled: true
  timestampField: updated_at   # which record field feeds the persisted state

query: |
  SELECT id, payload
  FROM events
  WHERE updated_at > $1
  ORDER BY updated_at
parameters:
  - state.lastRunTimestamp

On the first run, when no state is persisted yet, state.lastRunTimestamp is the epoch (1970-01-01T00:00:00Z) so the query returns everything. state.lastRunId is nil on the first run — coalesce a starting bound with state.lastRunId ?? 0.

Examples

examples/07-database-input-basic-to-http.yamlview source ↗
07-database-input-basic-to-http.yaml
name: database-input-basic-to-http
version: 1.0.0
description: Read rows from a database and send them as an HTTP batch.
tags:
  - database-input
  - http-output
input:
  type: database
  connectionStringRef: ${SOURCE_DATABASE_URL}
  driver: postgres
  query: |
    select id, email, updated_at
    from customers
    where active = true
    order by updated_at asc
filters:
  - type: mapping
    mappings:
      - source: id
        target: id
        transforms:
          - op: toString
      - source: email
        target: email
        transforms:
          - op: lowercase
      - source: updated_at
        target: updatedAt
        transforms:
          - op: dateFormat
            format: YYYY-MM-DDTHH:mm:ss
output:
  type: httpRequest
  endpoint: https://destination.example.com/api/customers/sync
  method: POST
  requestMode: batch
examples/08-database-input-limit-offset-to-database.yamlview source ↗
08-database-input-limit-offset-to-database.yaml
name: database-input-limit-offset-to-database
version: 1.0.0
description: Read a paginated SQL source and write rows into another database.
tags:
  - database-input
  - pagination
input:
  type: database
  connectionStringRef: ${SOURCE_DATABASE_URL}
  driver: postgres
  query: |
    select id, sku, quantity
    from inventory
    order by id asc
    limit $1 offset $2
  parameters:
    - pagination.limit
    - pagination.offset
  pagination:
    type: limit-offset
    limit: 500
filters:
  - type: set
    target: sync_source
    value: source-db
output:
  type: database
  connectionStringRef: ${WAREHOUSE_DATABASE_URL}
  driver: postgres
  query: |
    insert into inventory_snapshot (id, sku, quantity, sync_source)
    values ($1, $2, $3, $4)
  parameters:
    - record.id
    - record.sku
    - record.quantity
    - record.sync_source
  transaction: true
examples/09-database-input-cursor-incremental.yamlview source ↗
09-database-input-cursor-incremental.yaml
name: database-input-cursor-incremental
version: 1.0.0
description: Read database rows incrementally using cursor pagination metadata.
tags:
  - database-input
  - incremental
input:
  type: database
  connectionStringRef: ${SOURCE_DATABASE_URL}
  driver: postgres
  query: |
    select id, customer_id, total, updated_at
    from orders
    where updated_at > $1
      and id > $2
    order by id asc
    limit $3
  parameters:
    - state.lastRunTimestamp
    - pagination.cursor ?? state.lastRunId ?? 0
    - pagination.limit
  pagination:
    type: cursor
    limit: 200
    cursorField: id
  incremental:
    enabled: true
    timestampField: updated_at
    idField: id
filters:
  - type: mapping
    mappings:
      - source: id
        target: order.id
      - source: customer_id
        target: order.customerId
      - source: total
        target: order.total
        transforms:
          - op: toFloat
output:
  type: httpRequest
  endpoint: https://destination.example.com/api/orders
  method: POST
  requestMode: batch

Cross-references

State persistence

Persist cursor + timestamp between scheduled runs.

database output

Write to a database — the symmetric module.

sql_call filter

Per-record SQL enrichment between input and output.

rateLimit

rateLimit configuration for the webhook module.

pagination

pagination configuration for the database module.

On this page

Minimal exampleOptionsConnectionPaginationLimit + offsetCursorIncremental queriesExamplesCross-references