data_updater_v2.jobs.base

base.py — what every data_updater_v2 job shares, and nothing one job owns.

A job class subclasses DataJob. Its steps are its own methods (a step declared here would be every subclass’s, and two subclasses would collide on it) and its JOBS its own dict. What they share is how a step starts and what it leaves behind:

the database     common-db's ``DBClient`` from the environment (the framework's
                 ``_make_db``): the pipeline's database, with the central ``configs`` and
                 ``monitoring`` databases as its targets
its config row   in ``configs.data_updater_v2.<group>``, the table of its class's group
                 (``CONFIGS``; the dbt workloads': ``dbt``), one row a Dagster job:
                 ``enabled`` (the kill switch) and ``params`` (what its steps hand the
                 DataClient), read live at every step and parsed into the job's ``PARAMS``
the DataClient   one a step (``data``), on that database, owning its vendor clients
the session      the day a run is about (the ``SESSION_TAG`` a sensor stamps, else New
                 York's today)
the records      a write's receipt filed on its product (``landed``), a table found
                 current with nothing to write recorded too (``current``), a check's
                 issues opened and closed (``settle``)

Classes

DataJob

Bases: DGFlow

The base of every data_updater_v2 job class; no steps of its own.

Carried state, for one step execution (the framework builds the class fresh for each): db (the DBClient it owns and closes), params (its row’s, parsed) and monitoring (the run’s MonitoringRepo on the monitoring target). Nothing crosses steps but the tables.

assess

assess(dc: DataClient, windows: Sequence[tuple[Mapping[str, Any], Mapping[str, Any], Mapping[str, Any], Any]]) -> list[Any]

A backfill’s issues the sources could not move, closed as what they are. Each window is (what daily_backfill handed, the build’s result, its gaps, the write’s result). A listing’s days in a window are assessed when every source answered (no error gap for it) and the window landed nothing new for it: no bar at all, or only bars already stored (the sources stand by what they serve: the same invalid or odd bars again). An issue is closed source_assessment once, naming every assessed day of every window it spanned (a whole-history window names none: the issue whole), and is never raised again for them (daily_issues’ rule); the listing’s open fetch_failed goes with it. An issue a window did move, or a source did not answer for, stays open: the check re-states what remains. Returns what was closed.

current

current(dc: DataClient, product: str, message: str = 'read whole: nothing new or changed') -> str | None

Record that product is current for the run’s session though nothing was written: a step that read its source whole and found the stored table already holding it (a catalog with no era new or changed, a window with no line). A complete receipt with no rows, so what waits on the table’s landing (DataClient.last_landing: a prices sensor waiting on its catalog) sees the session landed. Never for a source unread: that step skips before. Returns the receipt id (None when the stage did not open).

data

data() -> AsyncIterator[DataClient]

A DataClient for this step on the job’s database, borrowed: it closes only what it owns (its vendor clients and limiter). The database’s async connections, opened on this step’s event loop, are closed on it when the block ends (aclose): once the loop is gone they cannot be (they reopen on the next use).

enabled_for_job

enabled_for_job(job: str) -> bool

The gate the Dagster daemon asks before a schedule or a sensor launches job (no setup runs there): its row’s enabled. A missing row raises, and the framework then skips the launch, naming why.

fetched

fetched(dc: DataClient, section: str, model: Any, gaps: Mapping[str, Mapping[str, Any]], rows: Sequence[Any]) -> dict[str, list[Any]]

A fetch’s failures opened as issues (fetch_failures), and the issues an earlier fetch opened on the table for the listings that answered this time closed: a table with no check of its own (the metrics’) settles its fetch failures this way, so an issue lives exactly as long as its listing goes unanswered.

landed

landed(dc: DataClient, product: str, written: Any, findings: Iterable[Mapping[str, Any]] = (), target_key: str | None = None, partial: bool | None = None) -> str | None

Record one write of product (a produces= name): its receipt on the table it landed in, with the finding counts, filed on the product so the product’s checks land on it. A write to another table than the product’s own (a scratch target) leaves the product unconfirmed. written None (no rows: nothing written) records nothing, and so does a step whose monitoring stage did not open. Returns the receipt id.

settle

settle(dc: DataClient, found: Mapping[str, Any]) -> dict[str, list[Any]]

A check’s issues (daily_issues, intraday_issues, …): what it observed opened, what no longer holds closed (the data proves it), and what an ended or unfetched era owed closed as superseded. Returns them, by what was done.

setup

setup(cfg: RunConfig) -> None

The job’s database, its config row read and parsed, the run’s monitoring. A disabled row ends the step as a named skip, before any vendor call.

teardown

teardown() -> None

The database closed — also after a setup that did not finish.

JobParams

Bases: BaseModel

A job’s row params: what its steps hand the DataClient. A job with none takes this model as it is (its row’s params empty). Unknown keys are refused, values typed and frozen.

Functions

config_group

config_group(job: str, own: str | None = None) -> str

The group whose table holds job’s row: its class’s CONFIGS as the census tells it ("dbt" for a dbt workload), else own (a class asking for a job of its own that no census lists). A job neither knows raises LookupError, as a missing row does.

job_enabled

job_enabled(db: Any, job: str) -> bool

A job’s kill switch as stored, for a dbt workload’s sensor and gate (no setup runs there): its row’s enabled. A missing row raises, as for every job.

row_params

row_params(db: Any, job: str) -> dict[str, Any]

A job’s row params as stored, for a sensor (no setup runs there): the probes it takes from the row, the first symbols of the set.

table_of

table_of(model: Any) -> str

A table model’s qualified name, as the issues and receipts name it.

unread

unread(section: str, model: Any, error: BaseException) -> dict[str, Any]

The issue a source unread opens on a table (one per table, whatever the source): the repair runs the step again, and the table landing closes it.