Read input data with a SQL client

Open a managed connection

Create connections through a pool and keep credentials outside source code. The adapter should own database-specific behavior and return normalized input envelopes to the runtime.

import os
import psycopg

with psycopg.connect(os.environ["DATABASE_URL"]) as connection:
    with connection.cursor() as cursor:
        cursor.execute(
            """
            SELECT event_id, event_type, occurred_at, workflow_id, payload
            FROM incoming_events
            WHERE occurred_at >= %s
            ORDER BY occurred_at, event_id
            LIMIT %s
            """,
            (since, 1000),
        )
        rows = cursor.fetchall()

Ingest rows safely

• Always bind parameters through the driver; never interpolate external values into SQL.
• Use a read-only database role and restrict it to the required schema and tables.
• Set statement and connection timeouts so a slow source cannot stall the ingestion worker.
• Page by stable indexed columns instead of OFFSET for repeatable, scalable reads.
• Commit the ingestion checkpoint only after the corresponding inputs are durably accepted.

Choose polling or change capture

Scheduled polling is appropriate for small or infrequent batches. For continuous updates, use the database change stream or an outbox table and preserve its sequence as the ingestion checkpoint.