Postgres table — `postgres`

Polls a table incrementally, one event per new/changed row. The cursor tracks a monotonic column (an autoincrement id, or updated_at); an entity column (e.g. tenant_id) becomes the key. Your application’s source-of-truth data in the timeline. Supports discovery: it reads information_schema to propose the cursor and key columns.

Prerequisites: the postgres extra (pip install 'navflow[postgres]') and a DSN — set the PG_DSN or DATABASE_URL environment variable on the daemon.

Set up

Agent

Ask a connected agent:

Use navflow to ingest the orders table, keyed by tenant_id.

The agent calls discover_source (reads information_schema, proposes the cursor and key columns), test_source (dry-run — fetches sample rows), then create_source. Walkthrough: Connect a Postgres table.

Console

Sources → Add source → postgres. Set:

table: e.g. orders or public.orders

Discover proposes cursor_column, cursor_type, and a key_column from the table’s schema; adjust and save.

YAML / API


- name: orders
  connector: postgres
  poll: 10s
  config:
    table: orders
    cursor_column: updated_at
    cursor_type: timestamp
    key_column: tenant_id
    labels:
      - { name: tenant_id, field: tenant_id, primary: true }
      - { name: status, field: status }

Or POST /api/sources with the same object.

Configuration

field	type	description
`table`*	string	table to poll, e.g. `orders` or `public.orders`
`cursor_column`*	string	monotonic column tracking new rows: an `id` or `updated_at`
`cursor_type`	string	`int` (autoincrement, the default) or `timestamp` (`updated_at`)
`key_column`	string	column whose value is the entity key, e.g. `tenant_id`
`time_column`	string	timestamp column for `event_time` (default: the cursor column if it is a timestamp, else ingest time)
`limit`	number	rows per poll (default 200)
`dsn`	string	`postgresql://user:pass@host:port/db` (prefer `PG_DSN`)
`text_template`	string	optional `str.format` template over the row, e.g. `'{tenant_id}: {status} ${amount}'`

Prefer the PG_DSN / DATABASE_URL environment variable over the dsn field — a DSN in the config is exported to the catalog YAML.

Verify

Open the source in the console: Recent events shows rows on the next poll; Fields reports each column’s coverage — use it to confirm the key column is reliably present.

Postgres table — postgres