PostgresSrc

class

class PostgresSrc(
    queries: QueriesSpec,
    transactional: bool = True,
    schema_overrides: SchemaOverrideSpec | None = None,
    initial_values: BasicDictSpec | None = None,
    src_cfg: SrcCfgSpec | None = None,
)

Bases: Src

Categories: source

Postgres source for @publisher -- one output slot per query.

Every table this source publishes automatically carries two metadata columns: @td.sql.query, the text of the query that produced the row, and @td.sql.initial_values, the JSON serialization of the bind values that query ran with (an empty JSON object when it has no binds). Setting the src_cfg key tabsdata.src_metadata.drop to True leaves them off.

Examples

Publish multiple Postgres tables, one output slot per query in order:

@publisher(
    source=PostgresSrc(
        queries=[
            "SELECT * FROM wld_country",
            "SELECT * FROM wld_city",
        ],
    ),
    output_tables=["pg_country", "pg_city"],
)
def ingest(country: TableFrameSpec, city: TableFrameSpec):
    return country, city

Incremental read: an initial_values bind seeds the :wm watermark on the first run, then ctx advances it each run:

@publisher(
    source=PostgresSrc(
        queries=[
            "SELECT * FROM wld_city WHERE id > :wm ORDER BY id LIMIT 10",
        ],
        initial_values={"wm": 0},
    ),
    output_tables=["pg_city_inc"],
)
def ingest_inc(city: TableFrameSpec, ctx: TrxCtx):
    if city is not None:
        row = city.max_for("id")
        if row is not None and row["id"] is not None:
            ctx.set_attr("wm", int(row["id"]))
    return city

Parameters

parameter

queriesQueriesSpec (list[str])

SELECT statements; one output slot per query, in order.

parameter

transactionalbool

When True (default), run all queries inside one transactional session.

parameter

schema_overridesSchemaOverrideSpec | None (list[dict[str, DataType] | None] | None)

Optional per-query column dtype hints. When given it must have exactly one entry per query (same length and order as queries); each entry is a {column: DataType} map or None.

parameter

initial_valuesBasicDictSpec | None (dict[str, str | int | float | bool] | None)

Optional {bind_name: value} seed dict. Each name is bound into the query as a SQLAlchemy :var parameter at execute time, and acts as the first-run seed: the plugin reads ctx.get_attr(bind_name) per bind and falls back to this value when the ctx store has no entry. Persist the next value with ctx.set_attr(bind_name, v).

parameter

src_cfgSrcCfgSpec | None (Mapping[Literal['postgres.logging_level', 'tabsdata.sql.read_engine', 'tabsdata.sql.chunk_size', 'tabsdata.src_metadata.drop'], Any] | None)

Optional config mapping over the allowed keys only (postgres.logging_level, tabsdata.sql.read_engine, tabsdata.sql.chunk_size, tabsdata.src_metadata.drop).

Methods

method

validate

def validate()

Check schema_overrides has one entry per query.

Called by the framework; raises PostgresValidateException when the lengths differ.