Skip to main content

v1

flow_specs record, v1 — canonical aliases.

Consumers and the record registry import from here and never touch the inner module names, mirroring runs/v1/__init__.py.

Module​

Submodules​

Classes​

Record​

class Record(**data: Any):

The persisted replay material for one linked (project, datasource).

Attributes

  • project_id: Project the datasource is linked to.
  • datasource_name: Datasource the project is linked to.
  • dag_name: The FlowSpec's name, which is also the DAG and Prefect flow name. Denormalised so lineage can be read without deserialising flow_spec_raw.
  • flow_spec_raw: The FlowSpec exactly as the Hub sent it.
  • origin_task_id: Task id of the triggering envelope, retained so a recovered attempt reports under the task the Hub already tracks. None for the CLI path, which has no envelope.
  • gave_up_at: When the lineage hit the unproductive-attempt bound. None while the lineage is still allowed to recover.
  • gave_up_reason: Support-facing explanation of why it gave up.
  • linked_at: When this link was last recorded.
  • tags: Arbitrary flat metadata.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static dag_name : str
  • static datasource_name : str
  • static gave_up_reason : str | None
  • static model_config
  • static origin_task_id : str | None
  • static project_id : str

  • gave_up : bool - Whether this lineage has been abandoned.

    Reads the timestamp rather than the reason, so a give-up recorded with no reason still counts.

  • inventory_count_at_last_run : int | None - How many indexed files this lineage's last completed run saw.

    None when no run has completed since the lineage was linked, which the growth trigger reads as "no baseline yet" and treats as due. That fails toward doing the work: the alternative — reading it as "nothing has grown" — would wedge a lineage whose every run so far has failed, leaving it to the daily cron for ever.

    A non-integer value reads as None for the same reason a corrupt last_rerun_at does: one hand-edited row should cost an extra rerun, not stall the trigger.

  • last_rerun_at : datetime.datetime | None - When this lineage was last rerun, by any rerun trigger.

    None when it has never been rerun, or when the stored value is not a parseable aware timestamp — a corrupt tag reads as "never rerun", which costs one extra rerun rather than stalling the lineage forever.

    A naive timestamp is rejected for that same reason, and it is the case that has to be rejected explicitly: fromisoformat accepts one happily, and every consumer compares the result against an aware datetime, which raises TypeError. That exception surfaces inside due_for_periodic_rerun, outside any per-record guard, so one hand-edited or restored row would take down reruns for every lineage on the pod. Everything this SDK writes is UTC-aware; a naive value came from somewhere else, and guessing its zone is not better than rerunning once more than needed.

  • last_rerun_trigger : str | None - Which rerun trigger last ran this lineage, if any.

  • reruns_since_completion : int - How many reruns have started since this lineage last completed a run.

    Incremented when a rerun takes its concurrency slot and reset when any background run of the lineage succeeds, so a non-zero value means that many runs began and none of them finished. The growth trigger's bound is on this: a lineage failing terminally banks no inventory count, and without a bound reads as "no baseline yet" on every poll for ever.

    Reruns rather than runs, because a slot is the thing being consumed — and a rerun that timed out waiting for one never acquired it, so it is correctly not counted as an attempt that achieved nothing.

    A missing or non-integer value reads as 0, the same direction every other tag here fails in: a corrupt row costs an extra rerun rather than standing the trigger down.

ORM​

class ORM(**kwargs):

SQLAlchemy model for the flow_specs cache table.

Holds the replay material for one (project_id, datasource_name): the FlowSpec exactly as the Hub sent it, plus the lineage bookkeeping recovery needs. Without this row a crashed background run cannot be rebuilt, because the FlowSpec exists nowhere else on the pod — it arrives inside the DATASET_PROJECT_LINKED message and was previously discarded once parsed.

One row per (project_id, datasource_name), upserted on every dataset-project link: the latest link wins, and re-linking is therefore the natural reset for a lineage that gave up.

The table also acts as the discriminator for recovery. Only the DATASET_PROJECT_LINKED path writes it, so "is this a background DAG run?" is answered by joining against this table rather than by pattern-matching the free-form type column in runs, which four different writers share.

Note the deliberate asymmetry with runs: this table describes a trigger and its lineage, not an execution. There is exactly one row per linked (project, datasource) however many runs that link produces.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance's class are allowed. These could be, for example, any mapped columns or relationships.

Ancestors​

Variables​

  • dag_name : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]
  • datasource_name : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]
  • flow_spec_raw : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]
  • gave_up_at : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]
  • gave_up_reason : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]
  • linked_at : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]
  • origin_task_id : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]
  • project_id : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]
  • tags : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]