Postgres

The Postgres connector lets Tabsdata run queries against Postgres and write tables into Postgres.

The PostgresSrc connector can be used by a publisher function to read data from Postgres into a Tabsdata table.

The PostgresDest connector can be used by a subscriber function to write data from a Tabsdata table into Postgres.

warning

The PostgreSQL connector needs the psycopg2-binary driver, which Tabsdata does not redistribute. See Installing Third-Party Database Drivers to add it to your Tabsdata server.

Publisher

The PostgresSrc connector can be used by a publisher function to read data from Postgres into a Tabsdata table. See the full publisher walkthrough

The publisher is registered in a collection created with a PostgresSrcConn connection. The connection, and how to create the collection, are described in Publisher connection, at the end of this page.

Prerequisites

• The Postgres server is reachable from the Tabsdata server.
• The connection user can run every query in queries and read every referenced table or view.

Publisher

Configure PostgresSrc as the source of a publisher function to select the queries that Tabsdata runs against Postgres.

publish_orders.py

from tabsdatak.api import publisher
from tabsdatak.conn.postgres import PostgresSrc

@publisher(
    source=PostgresSrc(
        queries=[
            "SELECT * FROM orders",
        ],
    ),
    output_tables=["orders"],
)
def publish_orders(orders):
    return orders

• required
• optional
• optional

Function Config Parameters

queries

Defines the SQL queries the publisher runs against Postgres.

Each element in queries represents one source slot and maps positionally to an argument in the publisher function.

For example:

PostgresSrc(
    queries=[
        "SELECT * FROM country",
        "SELECT * FROM city",
    ],
)

initial_values

The first-run values for named parameters in queries, written :name. On later runs each parameter takes the value the publisher stored with ctx.set_attr(name, value), so a query can read only the rows added since the last run.

Parameters bind values such as IDs and timestamps, not identifiers. Schema and table names have to be written into the query.

For example:

from tabsdatak.api import TableFrameSpec, TrxCtx, publisher
from tabsdatak.conn.postgres import PostgresSrc


@publisher(
    source=PostgresSrc(
        queries=["SELECT * FROM orders WHERE id > :last_id ORDER BY id"],
        initial_values={"last_id": 0},
    ),
    output_tables=["orders"],
)
def publish_orders(orders: TableFrameSpec, ctx: TrxCtx) -> TableFrameSpec:
    row = orders.max_for("id")
    if row is not None and row["id"] is not None:
        ctx.set_attr("last_id", int(row["id"]))
    return orders

src_cfg

Sets additional source configuration.

Subscriber

The PostgresDest connector can be used by a subscriber function to write data from a Tabsdata table into Postgres. See the full subscriber walkthrough

The subscriber is registered in a collection created with a PostgresDestConn connection. The connection, and how to create the collection, are described in Subscriber connection, at the end of this page.

Prerequisites

• The Postgres server is reachable from the Tabsdata server.
• The connection user can create missing destination tables and insert rows. With if_table_exists="replace", it can also truncate existing tables.

Subscriber

Configure PostgresDest as the destination of a subscriber function to define which tables Tabsdata writes in Postgres.

write_orders.py

from tabsdatak.api import subscriber
from tabsdatak.conn.postgres import PostgresDest

@subscriber(
    destination=PostgresDest(
        tables=[
            "orders",
        ],
    ),
    input_tables=["orders"],
)
def write_orders(orders):
    return orders

• required
• optional
• optional

Function Config Parameters

tables

Defines the destination tables the subscriber writes to in Postgres.

Each element in tables represents one destination slot and maps positionally to a value returned by the subscriber function.

For example:

PostgresDest(
    tables=[
        "vendors",
        "items",
    ],
)

if_table_exists

Controls what happens when the destination table already has rows. "append" (the default) keeps inserting into the existing rows. "replace" truncates the table before inserting. Neither mode changes the destination table's schema, so the returned frame's columns need to match the table Postgres already has.

dest_cfg

Sets additional destination configuration.

Query results

Each entry in queries produces one source slot. The slots map by position to the publisher function arguments and then to output_tables. Keep those three lists in the same order.

Queries run when the publisher is triggered. They are not a continuous change feed. To read incrementally, put named parameters such as :last_id in the query, set their first values with initial_values, and save the next values with ctx.set_attr() after each successful run.

Write behavior

Each entry in tables is one destination slot and maps by position to a value returned by the subscriber function. Returning None for a slot skips that write.

if_table_exists="append" keeps the rows already in the destination and inserts the returned rows. "replace" truncates the table before inserting. Neither mode changes an existing table's schema, so column additions, removals, and type changes need a database migration before the subscriber runs.

Operational notes

• Without trigger_by, a publisher runs only when triggered manually, while a subscriber runs when any input table commits. Set trigger_by to change when the function runs.
• Tabsdata checks the connection document when the collection is created, but it does not connect to Postgres then. A bad host, credential, query, or permission fails when the function runs.
• Every function in the collection uses the same connection. Use another collection when functions need another database or another set of credentials.

Publisher connection

Postgres publishers use PostgresSrcConn to define the database connection and credentials used by the publisher.

conn-postgres-in.yaml

kind: connectionDef
apiVersion: '1.0'
type: tabsdatak.conn.postgres:PostgresSrcConn
spec:
  uri: 'postgres://db.internal:5432/world'
  credentials:
    kind: userPasswordCredentials
    apiVersion: '1.0'
    type: tabsdatak.conn.common.types:UserPassword
    spec:
      user: 'secret:tabsdata_reader'
      password: 'secret:example-password'

• required
• required
• optional
• optional

Create the collection

A publisher is registered in a source collection, and a source collection is created with its connection. Every function registered in the collection uses that connection.

Generate a connection template, which lists all the connection fields:

tdk connection template --type postgres-sql-in --file conn-postgres-in.yaml

Edit the template file with the Postgres connection information, check it, and create the collection with it:

tdk connection validate --file conn-postgres-in.yaml

tdk collection create --name postgres_in --conn-file conn-postgres-in.yaml

Then register the publisher in the collection:

tdk fn register --coll postgres_in --path publish_orders.py::publish_orders

• The collection goes in the sources group, because the connection is a source connection. --group can be left out; if it is given, it must be sources.
• secret:<value> gives a secret value in the file itself. $secret:<ENV_VAR> reads it from an environment variable when the collection is created, so the value never appears in the file. Either way, Tabsdata stores the secret separately, and the stored connection only shows a placeholder.
• Functions in the collection must use the connector of its connection: PostgresSrc for a collection created with PostgresSrcConn. Transformers go in a collection of their own.
• To change the connection later, for example to rotate a password, run tdk collection update --name postgres_in --conn-file conn-postgres-in.yaml. The connector cannot change; to use another one, create another collection.
• Creating a source or destination collection requires permission to manage the project.

Connection Config Parameters

uri

The SQLAlchemy-style connection URI: postgres://host:port/database. Written as a plain value here, but accepts secret: too.

credentials

The database user and password, spliced into the URI at connect time. UserPassword requires a user and password.

entity_whitelist

A YAML list of regular expressions, not a comma-separated string. It is checked to be a list of valid regular expressions, but it has no effect yet. Leave it out.

conn_cfg

Driver-side options forwarded to the connection as SQLAlchemy connect_args.

Subscriber connection

Postgres subscribers use PostgresDestConn to define the database connection and credentials used by the subscriber.

conn-postgres-out.yaml

kind: connectionDef
apiVersion: '1.0'
type: tabsdatak.conn.postgres:PostgresDestConn
spec:
  uri: 'postgres://db.internal:5432/warehouse'
  credentials:
    kind: userPasswordCredentials
    apiVersion: '1.0'
    type: tabsdatak.conn.common.types:UserPassword
    spec:
      user: 'secret:tabsdata_reader'
      password: 'secret:example-password'

• required
• required
• optional
• optional

Create the collection

A subscriber is registered in a destination collection, and a destination collection is created with its connection. Every function registered in the collection uses that connection.

Generate a connection template, which lists all the connection fields:

tdk connection template --type postgres-sql-out --file conn-postgres-out.yaml

Edit the template file with the Postgres connection information, check it, and create the collection with it:

tdk connection validate --file conn-postgres-out.yaml

tdk collection create --name postgres_out --conn-file conn-postgres-out.yaml

Then register the subscriber in the collection:

tdk fn register --coll postgres_out --path write_orders.py::write_orders

• The collection goes in the destinations group, because the connection is a destination connection. --group can be left out; if it is given, it must be destinations.
• secret:<value> gives a secret value in the file itself. $secret:<ENV_VAR> reads it from an environment variable when the collection is created, so the value never appears in the file. Either way, Tabsdata stores the secret separately, and the stored connection only shows a placeholder.
• Functions in the collection must use the connector of its connection: PostgresDest for a collection created with PostgresDestConn. Transformers go in a collection of their own.
• To change the connection later, for example to rotate a password, run tdk collection update --name postgres_out --conn-file conn-postgres-out.yaml. The connector cannot change; to use another one, create another collection.
• Creating a source or destination collection requires permission to manage the project.

Connection Config Parameters

uri

The SQLAlchemy-style connection URI: postgres://host:port/database. Written as a plain value here, but accepts secret: too.

credentials

The database user and password, spliced into the URI at connect time. UserPassword requires a user and password.

entity_whitelist

A YAML list of regular expressions, not a comma-separated string. It is checked to be a list of valid regular expressions, but it has no effect yet. Leave it out.

conn_cfg

Driver-side options forwarded to the connection as SQLAlchemy connect_args.