• 主頁
• Airflow 公共介面
• airflow.models.dagrun

airflow.models.dagrun

屬性

CreatedTasks
RUN_ID_REGEX

類

TISchedulingDecision	DagRun.task_instance_scheduling_decisions 的返回型別。
DagRun	DAG 的呼叫例項。
DagRunNote	用於儲存關於 DagRun 例項的任意備註。

模組內容

airflow.models.dagrun.CreatedTasks

airflow.models.dagrun.RUN_ID_REGEX = '^(?:manual|scheduled|asset_triggered)__(?:\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\+00:00)$'

class airflow.models.dagrun.TISchedulingDecision

基類: NamedTuple

DagRun.task_instance_scheduling_decisions 的返回型別。

tis: list[airflow.models.taskinstance.TaskInstance]

schedulable_tis: list[airflow.models.taskinstance.TaskInstance]

changed_tis: bool

unfinished_tis: list[airflow.models.taskinstance.TaskInstance]

finished_tis: list[airflow.models.taskinstance.TaskInstance]¶

class airflow.models.dagrun.DagRun(dag_id=None, run_id=None, *, queued_at=NOTSET, logical_date=None, run_after=None, start_date=None, conf=None, state=None, run_type=None, creating_job_id=None, data_interval=None, triggered_by=None, backfill_id=None, bundle_version=None)

基類: airflow.models.base.Base, airflow.utils.log.logging_mixin.LoggingMixin

DAG 的呼叫例項。

DagRun 可以由排程器建立（即排程執行），也可以由外部觸發器建立（即手動執行）。

active_spans

__tablename__ = 'dag_run'

id

dag_id

queued_at

logical_date

start_date

end_date

run_id

creating_job_id

run_type

triggered_by

conf

data_interval_start

data_interval_end

run_after

last_scheduling_decision

log_template_id

updated_at

clear_number

backfill_id

此 DagRun 當前關聯的 backfill。

如果例如 DagRun 被清除以便重新執行，或者可能重新 backfill，則此關聯可能會改變。

bundle_version

scheduled_by_job_id

context_carrier

span_status

dag: airflow.models.dag.DAG | None

__table_args__

task_instances

task_instances_histories

dag_model

dag_run_note

backfill

backfill_max_active_runs

max_active_runs

note

DEFAULT_DAGRUNS_TO_EXAMINE

__repr__()

validate_run_id(key, run_id)

屬性 dag_versions: list[airflow.models.dag_version.DagVersion]

返回與此 DagRun 的任務例項 (TI) 關聯的 DAG 版本。

屬性 version_number: int | None

返回與此 DagRun 最新任務例項 (TI) 關聯的 DAG 版本號。

check_version_id_exists_in_dr(dag_version_id, session=NEW_SESSION)

屬性 stats_tags: dict[str, str]

類方法 set_active_spans(active_spans)

get_state()

set_state(state)

改變 DagRun 的狀態。

根據下表實現屬性更改（行代表舊狀態，列代表新狀態）

狀態轉換矩陣

	QUEUED	RUNNING	SUCCESS	FAILED
None	queued_at = timezone.utcnow()	if empty: start_date = timezone.utcnow() end_date = None	end_date = timezone.utcnow()	end_date = timezone.utcnow()
QUEUED	queued_at = timezone.utcnow()	if empty: start_date = timezone.utcnow() end_date = None	end_date = timezone.utcnow()	end_date = timezone.utcnow()
RUNNING	queued_at = timezone.utcnow() start_date = None end_date = None		end_date = timezone.utcnow()	end_date = timezone.utcnow()
SUCCESS	queued_at = timezone.utcnow() start_date = None end_date = None	start_date = timezone.utcnow() end_date = None
FAILED	queued_at = timezone.utcnow() start_date = None end_date = None	start_date = timezone.utcnow() end_date = None

屬性 state

refresh_from_db(session=NEW_SESSION)

從資料庫重新載入當前的 dagrun。

引數:

session (sqlalchemy.orm.Session) – 資料庫會話

類方法 find(dag_id=None, run_id=None, logical_date=None, state=None, no_backfills=False, run_type=None, session=NEW_SESSION, logical_start_date=None, logical_end_date=None)

返回符合給定搜尋條件的 DagRun 集合。

引數:

• dag_id (str | list[str] | None) – 用於查詢 DAG 執行的 dag_id 或 dag_id 列表

• run_id (collections.abc.Iterable[str] | None) – 定義此 DAG 執行的執行 ID

• run_type (airflow.utils.types.DagRunType | None) – DAG 執行的型別

• logical_date (datetime.datetime | collections.abc.Iterable[datetime.datetime] | None) – 邏輯日期

• state (airflow.utils.state.DagRunState | None) – DAG 執行的狀態

• no_backfills (bool) – 不返回回填（True），返回所有（False）。預設為 False

• session (sqlalchemy.orm.Session) – 資料庫會話

• logical_start_date (datetime.datetime | None) – 從此日期開始執行的 DAG 執行

• logical_end_date (datetime.datetime | None) – 在此日期之前執行的 DAG 執行

classmethod find_duplicate(dag_id, run_id, *, session=NEW_SESSION)

返回指定 run_id 的 DAG 現有執行。

如果未找到此類 DAG 執行，則返回 None。

引數:

• dag_id (str) – 用於查詢重複項的 dag_id

• run_id (str) – 定義此 DAG 執行的執行 ID

• session (sqlalchemy.orm.Session) – 資料庫會話

static generate_run_id(*, run_type, logical_date=None, run_after)

基於執行型別、run_after 和邏輯日期生成執行 ID。

引數:

• run_type (airflow.utils.types.DagRunType) – DAG 執行型別

• logical_date (datetime.datetime | None) – 邏輯日期

• run_after (datetime.datetime) – DAG 執行不會在此日期之前啟動的日期。

static fetch_task_instances(dag_id=None, run_id=None, task_ids=None, state=None, session=NEW_SESSION)

返回此 DAG 執行的任務例項。

get_task_instances(state=None, session=NEW_SESSION)

返回此 DAG 執行的任務例項。

重定向到 DagRun.fetch_task_instances 方法。保留此方法是因為它在程式碼中廣泛使用。

get_task_instance(task_id, session=NEW_SESSION, *, map_index=-1)

返回此 DAG 執行中由 task_id 指定的任務例項。

引數:

• task_id (str) – 任務 ID

• session (sqlalchemy.orm.Session) – Sqlalchemy ORM Session

static fetch_task_instance(dag_id, dag_run_id, task_id, session=NEW_SESSION, map_index=-1)

返回此 DAG 執行中由 task_id 指定的任務例項。

引數:

• dag_id (str) – DAG ID

• dag_run_id (str) – DAG 執行 ID

• task_id (str) – 任務 ID

• session (sqlalchemy.orm.Session) – Sqlalchemy ORM Session

get_dag()

返回與此 DagRun 相關聯的 Dag。

返回:

DAG

返回型別:

airflow.models.dag.DAG

static get_previous_dagrun(dag_run, state=None, session=NEW_SESSION)

返回上一個 DagRun，如果存在。

引數:

• dag_run (DagRun) – Dag 執行

• session (sqlalchemy.orm.Session) – SQLAlchemy ORM Session

• state (airflow.utils.state.DagRunState | None) – Dag 執行狀態

static get_previous_scheduled_dagrun(dag_run_id, session=NEW_SESSION)

返回上一個 SCHEDULED DagRun，如果存在。

引數:

• dag_run_id (int) – DAG 執行 ID

• session (sqlalchemy.orm.Session) – SQLAlchemy ORM Session

set_dagrun_span_attrs(span)

start_dr_spans_if_needed(tis)

end_dr_span_if_needed()

update_state(session=NEW_SESSION, execute_callbacks=True)

根據其任務例項的狀態確定 DagRun 的總體狀態。

引數:

• session (sqlalchemy.orm.Session) – Sqlalchemy ORM Session

• execute_callbacks (bool) – 是否應直接呼叫 DAG 回撥（成功/失敗、SLA 等）（預設: true），或將其記錄為 returned_callback 屬性中的待處理請求

返回:

包含可在當前迴圈中排程的 tis 以及需要執行的 returned_callback 的元組

返回型別:

tuple[list[airflow.models.taskinstance.TaskInstance], airflow.callbacks.callback_requests.DagCallbackRequest | None]

task_instance_scheduling_decisions(session=NEW_SESSION)

notify_dagrun_state_changed(msg='')

verify_integrity(*, session=NEW_SESSION, dag_version_id=None)

透過檢查已移除的任務或尚未在資料庫中的任務來驗證 DagRun。

它將把狀態設定為 removed 或在需要時新增任務。

引數:

• dag_version_id (sqlalchemy_utils.UUIDType | None) – DAG 版本 ID

• session (sqlalchemy.orm.Session) – Sqlalchemy ORM Session

classmethod get_latest_runs(session=NEW_SESSION)

返回每個 DAG 的最新 DagRun。

schedule_tis(schedulable_tis, session=NEW_SESSION, max_tis_per_query=None)

將給定的任務例項設定為 scheduled 狀態。

schedulable_tis 的每個元素都應該已經設定了其 task 屬性。

任何沒有回撥或輸出埠的 EmptyOperator 將直接設定為 success 狀態。

所有 TIs 都應屬於此 DagRun，但這段程式碼位於熱路徑中，不會對此進行檢查——呼叫者有責任只使用來自單個 DAG 執行的 TIs 呼叫此函式。

get_log_template(*, session=NEW_SESSION)

class airflow.models.dagrun.DagRunNote(content, user_id=None)

基類: airflow.models.base.Base

用於儲存關於 DagRun 例項的任意備註。

__tablename__ = 'dag_run_note'

user_id

dag_run_id

content

created_at

updated_at

dag_run

__table_args__

__repr__()

上一頁 下一頁

此條目有幫助嗎？