OracleSrc

class

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

Bases: Src

Categories: source

Oracle 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 Oracle tables, one output slot per query in order:

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

Incremental read: an initial_values bind seeds the :wm watermark, FETCH FIRST n ROWS ONLY pages the scan, and ctx advances it each run. Oracle folds unquoted identifiers to uppercase, so the watermark column is read as ID:

@publisher(
    source=OracleSrc(
        queries=[
            "SELECT * FROM wld_city WHERE id > :wm "
            "ORDER BY id FETCH FIRST 10 ROWS ONLY"
        ],
        initial_values={"wm": 0},
    ),
    output_tables=["or_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['oracle.logging_level', 'tabsdata.sql.read_engine', 'tabsdata.sql.chunk_size', 'tabsdata.src_metadata.drop'], Any] | None)

Optional config mapping over the allowed keys only (oracle.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 OracleValidateException when the lengths differ.