MSSQLSrc

class

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

Bases: Src

Categories: source

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

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

Incremental read -- SQL Server has no LIMIT, so page with OFFSET ... FETCH NEXT (which needs ORDER BY); an initial_values bind seeds :wm on the first run and ctx advances it each run:

@publisher(
    source=MSSQLSrc(
        queries=[
            "SELECT * FROM wld_city WHERE id > :wm "
            "ORDER BY id OFFSET 0 ROWS FETCH NEXT 10 ROWS ONLY"
        ],
        initial_values={"wm": 0},
    ),
    output_tables=["ms_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 entry.

parameter

transactionalbool

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

parameter

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

Optional per-query column dtype hints; when given, must have the same length as queries.

parameter

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

Optional {bind_name: value} seed dict. Each name binds into the query as a SQLAlchemy :var parameter; it seeds the first run only, after which the plugin reads ctx.get_attr(bind_name) and the function persists new values with ctx.set_attr(bind_name, ...).

parameter

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

Optional connector config; supported keys are mssql.logging_level, tabsdata.sql.read_engine, tabsdata.sql.chunk_size and tabsdata.src_metadata.drop.

Methods

method

validate

def validate()

Cross-field validation: schema_overrides length must match queries.

Called by the framework; raises MSSQLValidateException.