Database Source

The Database source runs a SQL query on a recurring schedule and feeds each result row into the channel pipeline. After the query executes, an optional post_process_statement can mark rows as processed so they are not picked up again on the next poll.

Configuration

source:
  type: database
  driver: postgres
  dsn: postgres://${DB_USER}:${DB_PASS}@db.hospital.local:5432/orders?sslmode=require
  poll_interval: 15s
  query: >
    SELECT id, patient_id, medication, dosage, ordered_at
    FROM medication_orders
    WHERE processed = false
    ORDER BY ordered_at ASC
    LIMIT 100
  post_process_statement: >
    UPDATE medication_orders
    SET processed = true
    WHERE id = :id

Properties

Property	Type	Required	Default	Description
driver	string	Yes	—	Database driver: postgres, mysql, mssql, or oracle.
dsn	string	Yes	—	Data Source Name (connection string). Format varies by driver.
poll_interval	string	No	30s	How often to execute the query. Supports s, m, h suffixes.
query	string	Yes	—	SQL SELECT statement to retrieve unprocessed rows.
post_process_statement	string	No	—	SQL statement executed per row after successful processing. Use :column placeholders to reference result columns.

Supported Drivers

Driver	DSN Format
postgres	postgres://user:pass@host:5432/dbname?sslmode=require
mysql	mysql://user:pass@tcp(host:3306)/dbname?tls=true
mssql	sqlserver://user:pass@host:1433?database=dbname&encrypt=true
oracle	oracle://user:pass@host:1521/service_name

Complete Example

Poll a medication_orders table and forward new orders to a pharmacy system:

channel.yaml

channel:
  name: medication-order-poll
  description: Poll new medication orders and send to pharmacy

source:
  type: database
  driver: postgres
  dsn: postgres://${DB_USER}:${DB_PASS}@db.hospital.local:5432/orders?sslmode=require
  poll_interval: 15s
  query: >
    SELECT id, patient_id, medication, dosage, ordered_at
    FROM medication_orders
    WHERE processed = false
    ORDER BY ordered_at ASC
    LIMIT 100
  post_process_statement: >
    UPDATE medication_orders
    SET processed = true
    WHERE id = :id

transformer:
  type: typescript
  file: ./transforms/order-to-hl7.ts

destination:
  type: tcp
  host: pharmacy.hospital.local
  port: 2575
  mode: mllp

Caution

Always include a WHERE clause that limits the query to unprocessed rows (e.g. WHERE processed = false). Without it, the source will re-emit every row on every poll cycle, leading to duplicate messages and unnecessary load on the database.

Tip

Use LIMIT in your query to cap the number of rows per poll cycle. This prevents a large backlog from overwhelming downstream systems and keeps memory usage predictable.