You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ds...@apache.org on 2025/04/13 13:24:35 UTC
(airflow) branch main updated: Remove dag_version as a create_dagrun argument (#49148)
This is an automated email from the ASF dual-hosted git repository.
dstandish pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 0c883c5be7c Remove dag_version as a create_dagrun argument (#49148)
0c883c5be7c is described below
commit 0c883c5be7c6d24768b4f54904e582b4262da4b6
Author: Daniel Standish <15...@users.noreply.github.com>
AuthorDate: Sun Apr 13 06:24:28 2025 -0700
Remove dag_version as a create_dagrun argument (#49148)
I checked all the uses, and in every case (except one test), either "the latest dag version" was supplied, or None was supplied (in which case, the latest dag version is looked up and used anyway).
If we ever need to create a dag run with something other than the latest dag run version, then we can add it back.
Having this not be an argument makes it easier to control the behavior at dag run creation time.
---
airflow-core/src/airflow/api/common/trigger_dag.py | 3 -
.../api_fastapi/core_api/routes/public/assets.py | 2 -
.../api_fastapi/core_api/routes/public/dag_run.py | 2 -
.../src/airflow/cli/commands/task_command.py | 1 -
.../src/airflow/jobs/scheduler_job_runner.py | 6 -
airflow-core/src/airflow/models/backfill.py | 3 -
airflow-core/src/airflow/models/dag.py | 7 +-
airflow-core/src/airflow/models/dag_version.py | 14 +-
airflow-core/tests/unit/api_fastapi/conftest.py | 2 -
.../tests/unit/cli/commands/test_task_command.py | 3 -
airflow-core/tests/unit/jobs/test_scheduler_job.py | 155 +++++----------------
airflow-core/tests/unit/models/test_dag.py | 3 -
airflow-core/tests/unit/models/test_dagrun.py | 2 -
.../tests/unit/models/test_taskinstance.py | 14 +-
dev/airflow_perf/scheduler_dag_execution_timing.py | 1 -
devel-common/src/tests_common/pytest_plugin.py | 2 -
.../standard/sensors/test_external_task_sensor.py | 1 -
17 files changed, 54 insertions(+), 167 deletions(-)
diff --git a/airflow-core/src/airflow/api/common/trigger_dag.py b/airflow-core/src/airflow/api/common/trigger_dag.py
index 8378876f676..28ab45975d7 100644
--- a/airflow-core/src/airflow/api/common/trigger_dag.py
+++ b/airflow-core/src/airflow/api/common/trigger_dag.py
@@ -24,7 +24,6 @@ from typing import TYPE_CHECKING
from airflow.exceptions import DagNotFound, DagRunAlreadyExists
from airflow.models import DagBag, DagModel, DagRun
-from airflow.models.dag_version import DagVersion
from airflow.utils import timezone
from airflow.utils.session import NEW_SESSION, provide_session
from airflow.utils.state import DagRunState
@@ -104,7 +103,6 @@ def _trigger_dag(
run_conf = None
if conf:
run_conf = conf if isinstance(conf, dict) else json.loads(conf)
- dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
dag_run = dag.create_dagrun(
run_id=run_id,
logical_date=coerced_logical_date,
@@ -113,7 +111,6 @@ def _trigger_dag(
conf=run_conf,
run_type=DagRunType.MANUAL,
triggered_by=triggered_by,
- dag_version=dag_version,
state=DagRunState.QUEUED,
session=session,
)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
index 9a5ca3bc6cb..f2287206664 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
@@ -70,7 +70,6 @@ from airflow.models.asset import (
TaskOutletAssetReference,
)
from airflow.models.dag import DAG
-from airflow.models.dag_version import DagVersion
from airflow.utils import timezone
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunTriggeredByType, DagRunType
@@ -331,7 +330,6 @@ def materialize_asset(
run_after=run_after,
run_type=DagRunType.MANUAL,
triggered_by=DagRunTriggeredByType.REST_API,
- dag_version=DagVersion.get_latest_version(dag_id, session=session),
state=DagRunState.QUEUED,
session=session,
)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
index 8c701a9665d..985a3934f5c 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
@@ -72,7 +72,6 @@ from airflow.api_fastapi.logging.decorators import action_logging
from airflow.exceptions import ParamValidationError
from airflow.listeners.listener import get_listener_manager
from airflow.models import DAG, DagModel, DagRun
-from airflow.models.dag_version import DagVersion
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunTriggeredByType, DagRunType
@@ -411,7 +410,6 @@ def trigger_dag_run(
conf=params["conf"],
run_type=DagRunType.MANUAL,
triggered_by=DagRunTriggeredByType.REST_API,
- dag_version=DagVersion.get_latest_version(dag.dag_id),
state=DagRunState.QUEUED,
session=session,
)
diff --git a/airflow-core/src/airflow/cli/commands/task_command.py b/airflow-core/src/airflow/cli/commands/task_command.py
index 258ea5d6f91..de0120c7352 100644
--- a/airflow-core/src/airflow/cli/commands/task_command.py
+++ b/airflow-core/src/airflow/cli/commands/task_command.py
@@ -139,7 +139,6 @@ def _get_dag_run(
run_after=run_after,
run_type=DagRunType.MANUAL,
triggered_by=DagRunTriggeredByType.CLI,
- dag_version=None,
state=DagRunState.RUNNING,
session=session,
)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 43fc54386c5..8a84855e291 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -1462,8 +1462,6 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id)
continue
- latest_dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
-
data_interval = dag.get_next_data_interval(dag_model)
# Explicitly check if the DagRun already exists. This is an edge case
# where a Dag Run is created but `DagModel.next_dagrun` and `DagModel.next_dagrun_create_after`
@@ -1486,7 +1484,6 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
run_after=dag_model.next_dagrun_create_after,
run_type=DagRunType.SCHEDULED,
triggered_by=DagRunTriggeredByType.TIMETABLE,
- dag_version=latest_dag_version,
state=DagRunState.QUEUED,
creating_job_id=self.job.id,
session=session,
@@ -1535,8 +1532,6 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
)
continue
- latest_dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
-
triggered_date = triggered_dates[dag.dag_id]
cte = (
select(func.max(DagRun.run_after).label("previous_dag_run_run_after"))
@@ -1569,7 +1564,6 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
run_after=triggered_date,
run_type=DagRunType.ASSET_TRIGGERED,
triggered_by=DagRunTriggeredByType.ASSET,
- dag_version=latest_dag_version,
state=DagRunState.QUEUED,
creating_job_id=self.job.id,
session=session,
diff --git a/airflow-core/src/airflow/models/backfill.py b/airflow-core/src/airflow/models/backfill.py
index 1dc57a37ae6..2ca2e87396c 100644
--- a/airflow-core/src/airflow/models/backfill.py
+++ b/airflow-core/src/airflow/models/backfill.py
@@ -43,7 +43,6 @@ from sqlalchemy_jsonfield import JSONField
from airflow.exceptions import AirflowException, DagNotFound
from airflow.models.base import Base, StringID
-from airflow.models.dag_version import DagVersion
from airflow.settings import json
from airflow.utils import timezone
from airflow.utils.session import create_session
@@ -327,7 +326,6 @@ def _create_backfill_dag_run(
)
return
- dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
try:
dr = dag.create_dagrun(
run_id=DagRun.generate_run_id(
@@ -339,7 +337,6 @@ def _create_backfill_dag_run(
conf=dag_run_conf,
run_type=DagRunType.BACKFILL_JOB,
triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
state=DagRunState.QUEUED,
start_date=timezone.utcnow(),
backfill_id=backfill_id,
diff --git a/airflow-core/src/airflow/models/dag.py b/airflow-core/src/airflow/models/dag.py
index 665210331a5..19717623303 100644
--- a/airflow-core/src/airflow/models/dag.py
+++ b/airflow-core/src/airflow/models/dag.py
@@ -256,7 +256,6 @@ def _create_orm_dagrun(
conf: Any,
state: DagRunState | None,
run_type: DagRunType,
- dag_version: DagVersion | None,
creating_job_id: int | None,
backfill_id: int | None,
triggered_by: DagRunTriggeredByType,
@@ -290,8 +289,7 @@ def _create_orm_dagrun(
run.dag = dag
# create the associated task instances
# state is None at the moment of creation
- if not dag_version:
- dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
+ dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
run.verify_integrity(session=session, dag_version_id=dag_version.id if dag_version else None)
return run
@@ -1793,7 +1791,6 @@ class DAG(TaskSDKDag, LoggingMixin):
conf: dict | None = None,
run_type: DagRunType,
triggered_by: DagRunTriggeredByType,
- dag_version: DagVersion | None = None,
state: DagRunState,
start_date: datetime | None = None,
creating_job_id: int | None = None,
@@ -1867,7 +1864,6 @@ class DAG(TaskSDKDag, LoggingMixin):
conf=conf,
state=state,
run_type=run_type,
- dag_version=dag_version,
creating_job_id=creating_job_id,
backfill_id=backfill_id,
triggered_by=triggered_by,
@@ -2622,7 +2618,6 @@ def _get_or_create_dagrun(
run_type=DagRunType.MANUAL,
state=DagRunState.RUNNING,
triggered_by=triggered_by,
- dag_version=DagVersion.get_latest_version(dag.dag_id, session=session),
start_date=start_date or logical_date,
session=session,
)
diff --git a/airflow-core/src/airflow/models/dag_version.py b/airflow-core/src/airflow/models/dag_version.py
index 2b6483c9d83..529d5b7d4ee 100644
--- a/airflow-core/src/airflow/models/dag_version.py
+++ b/airflow-core/src/airflow/models/dag_version.py
@@ -112,18 +112,24 @@ class DagVersion(Base):
return dag_version
@classmethod
- def _latest_version_select(cls, dag_id: str) -> Select:
+ def _latest_version_select(cls, dag_id: str, bundle_version: str | None = None) -> Select:
"""
Get the select object to get the latest version of the DAG.
:param dag_id: The DAG ID.
:return: The select object.
"""
- return select(cls).where(cls.dag_id == dag_id).order_by(cls.created_at.desc()).limit(1)
+ query = select(cls).where(cls.dag_id == dag_id)
+ if bundle_version:
+ query = query.where(cls.bundle_version == bundle_version)
+ query = query.order_by(cls.created_at.desc()).limit(1)
+ return query
@classmethod
@provide_session
- def get_latest_version(cls, dag_id: str, *, session: Session = NEW_SESSION) -> DagVersion | None:
+ def get_latest_version(
+ cls, dag_id: str, *, bundle_version: str | None = None, session: Session = NEW_SESSION
+ ) -> DagVersion | None:
"""
Get the latest version of the DAG.
@@ -131,7 +137,7 @@ class DagVersion(Base):
:param session: The database session.
:return: The latest version of the DAG or None if not found.
"""
- return session.scalar(cls._latest_version_select(dag_id))
+ return session.scalar(cls._latest_version_select(dag_id, bundle_version=bundle_version))
@classmethod
@provide_session
diff --git a/airflow-core/tests/unit/api_fastapi/conftest.py b/airflow-core/tests/unit/api_fastapi/conftest.py
index 165b23f3531..942d0f76617 100644
--- a/airflow-core/tests/unit/api_fastapi/conftest.py
+++ b/airflow-core/tests/unit/api_fastapi/conftest.py
@@ -27,7 +27,6 @@ from fastapi.testclient import TestClient
from airflow.api_fastapi.app import create_app
from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser
from airflow.models import Connection
-from airflow.models.dag_version import DagVersion
from airflow.models.serialized_dag import SerializedDagModel
from airflow.providers.standard.operators.empty import EmptyOperator
@@ -161,7 +160,6 @@ def make_dag_with_multiple_versions(dag_maker, configure_git_connection_for_dag_
dag_maker.create_dagrun(
run_id=f"run{version_number}",
logical_date=datetime.datetime(2020, 1, version_number, tzinfo=datetime.timezone.utc),
- dag_version=DagVersion.get_version(dag_id=dag_id, version_number=version_number),
)
dag.sync_to_db()
diff --git a/airflow-core/tests/unit/cli/commands/test_task_command.py b/airflow-core/tests/unit/cli/commands/test_task_command.py
index b3dd69912ce..dbd60b091df 100644
--- a/airflow-core/tests/unit/cli/commands/test_task_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_task_command.py
@@ -101,7 +101,6 @@ class TestCliTasks:
logical_date=DEFAULT_DATE,
data_interval=data_interval,
run_after=DEFAULT_DATE,
- dag_version=None,
triggered_by=DagRunTriggeredByType.TEST,
)
@@ -353,7 +352,6 @@ class TestCliTasks:
data_interval=data_interval,
run_after=default_date2,
run_type=DagRunType.MANUAL,
- dag_version=None,
triggered_by=DagRunTriggeredByType.CLI,
)
ti2 = TaskInstance(task2, run_id=dagrun.run_id)
@@ -438,7 +436,6 @@ class TestLogsfromTaskRunCommand:
start_date=timezone.utcnow(),
state=State.RUNNING,
run_type=DagRunType.MANUAL,
- dag_version=None,
triggered_by=DagRunTriggeredByType.TEST,
)
self.tis = self.dr.get_task_instances()
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index b1119628e22..1f4b90e5f97 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -156,7 +156,6 @@ def create_dagrun(session):
state=state,
start_date=start_date,
triggered_by=DagRunTriggeredByType.TEST,
- dag_version=DagVersion.get_latest_version(dag.dag_id),
)
return _create_dagrun
@@ -383,8 +382,7 @@ class TestSchedulerJob:
with dag_maker(dag_id=dag_id, fileloc="/test_path1/") as dag:
EmptyOperator(task_id=task_id_1, on_failure_callback=lambda x: print("hi"))
- dagv = DagVersion.get_latest_version(dag.dag_id)
- dr = dag_maker.create_dagrun(dag_version=dagv)
+ dr = dag_maker.create_dagrun()
ti1 = dr.task_instances[0]
mock_stats_incr.reset_mock()
@@ -4287,7 +4285,6 @@ class TestSchedulerJob:
)
data_interval = dag.infer_automated_data_interval(DEFAULT_LOGICAL_DATE)
- dag_version = DagVersion.get_latest_version(dag.dag_id)
run1 = dag.create_dagrun(
run_id="test1",
run_type=DagRunType.SCHEDULED,
@@ -4297,7 +4294,6 @@ class TestSchedulerJob:
session=session,
data_interval=data_interval,
run_after=DEFAULT_LOGICAL_DATE,
- dag_version=dag_version,
triggered_by=DagRunTriggeredByType.TEST,
)
@@ -4313,7 +4309,6 @@ class TestSchedulerJob:
session=session,
data_interval=data_interval,
run_after=logical_date_2,
- dag_version=dag_version,
triggered_by=DagRunTriggeredByType.TEST,
)
@@ -4504,8 +4499,7 @@ class TestSchedulerJob:
BashOperator(task_id="dummy3", bash_command="true")
session = settings.Session()
- dag_version = DagVersion.get_latest_version(dag.dag_id)
- dag_run = dag_maker.create_dagrun(state=State.QUEUED, session=session, dag_version=dag_version)
+ dag_run = dag_maker.create_dagrun(state=State.QUEUED, session=session)
DAG.bulk_write_to_db("testing", None, [dag], session=session) # Update the date fields
@@ -4547,16 +4541,11 @@ class TestSchedulerJob:
schedule=timedelta(hours=1),
max_active_runs=1,
catchup=True,
- ) as dag:
+ ):
EmptyOperator(task_id="mytask")
- dag_version = DagVersion.get_latest_version(dag.dag_id)
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
for _ in range(29):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
# Explicitly set catchup=True as test specifically expects historical dates to be respected
with dag_maker(
@@ -4564,16 +4553,11 @@ class TestSchedulerJob:
start_date=timezone.datetime(2020, 1, 1),
schedule=timedelta(hours=1),
catchup=True,
- ) as dag2:
+ ):
EmptyOperator(task_id="mytask")
- dag_version = DagVersion.get_latest_version(dag2.dag_id)
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
for _ in range(9):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
scheduler_job = Job(executor=MockExecutor(do_update=False))
self.job_runner = SchedulerJobRunner(job=scheduler_job)
@@ -4603,30 +4587,20 @@ class TestSchedulerJob:
) as dag:
EmptyOperator(task_id="mytask")
dag1_dag_id = dag.dag_id
- dag_version = DagVersion.get_latest_version(dag1_dag_id)
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
for _ in range(29):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
with dag_maker(
"test_dag2",
start_date=timezone.datetime(2020, 1, 1),
schedule=timedelta(days=1),
catchup=True,
- ) as dag:
+ ):
EmptyOperator(task_id="mytask")
- dag_version = DagVersion.get_latest_version(dag.dag_id)
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
for _ in range(9):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
scheduler_job = Job(executor=MockExecutor(do_update=False))
self.job_runner = SchedulerJobRunner(job=scheduler_job)
@@ -4712,15 +4686,10 @@ class TestSchedulerJob:
) as dag:
EmptyOperator(task_id="mytask")
dag1_dag_id = dag.dag_id
- dag_version = DagVersion.get_latest_version(dag1_dag_id)
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
# Fewer DAG runs since we're only testing recent dates with catchup=False
for _ in range(2):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
with dag_maker(
"test_dag2",
@@ -4729,14 +4698,9 @@ class TestSchedulerJob:
catchup=False,
) as dag:
EmptyOperator(task_id="mytask")
- dag_version = DagVersion.get_latest_version(dag.dag_id)
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
for _ in range(2):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
scheduler_job = Job(executor=MockExecutor(do_update=False))
self.job_runner = SchedulerJobRunner(job=scheduler_job)
@@ -4846,31 +4810,21 @@ class TestSchedulerJob:
)
dag1_non_b_running, dag1_b_running, total_running = _running_counts()
- dag_version = DagVersion.get_latest_version(dag1_dag_id)
# now let's create some "normal" dag runs and verify that they can run
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
for _ in range(29):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
with dag_maker(
"test_dag2",
start_date=timezone.datetime(2020, 1, 1),
schedule=timedelta(days=1),
catchup=True,
- ) as dag2:
+ ):
EmptyOperator(task_id="mytask")
- dag_version = DagVersion.get_latest_version(dag2.dag_id)
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
for _ in range(9):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
# initial state -- nothing is running
assert dag1_non_b_running == 0
@@ -4971,34 +4925,24 @@ class TestSchedulerJob:
)
dag1_non_b_running, dag1_b_running, total_running = _running_counts()
- dag_version = DagVersion.get_latest_version(dag1_dag_id)
# Create fewer DAG runs since we're only testing recent dates with catchup=False
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
# With catchup=False, we only create a few runs instead of 29
for _ in range(4):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
with dag_maker(
"test_dag2",
start_date=timezone.datetime(2020, 1, 1),
schedule=timedelta(days=1),
catchup=False,
- ) as dag2:
+ ):
EmptyOperator(task_id="mytask")
- dag_version = DagVersion.get_latest_version(dag2.dag_id)
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
# With catchup=False, we only create a few runs instead of 9
for _ in range(2):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
# initial state -- nothing is running
assert dag1_non_b_running == 0
@@ -5115,31 +5059,21 @@ class TestSchedulerJob:
assert session.scalar(select(func.count()).select_from(DagRun)) == 6
assert session.scalar(select(func.count()).where(DagRun.dag_id == dag1_dag_id)) == 6
- dag_version = DagVersion.get_latest_version(dag1_dag_id)
# now let's create some "normal" dag runs and verify that they can run
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
for _ in range(29):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
with dag_maker(
"test_dag2",
start_date=timezone.datetime(2020, 1, 1),
schedule=timedelta(days=1),
catchup=True,
- ) as dag2:
+ ):
EmptyOperator(task_id="mytask")
- dag_version = DagVersion.get_latest_version(dag2.dag_id)
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
for _ in range(9):
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
# ok at this point, there are new dag runs created, but no new running runs
dag1_non_b_running, dag1_b_running, total_running = _running_counts()
@@ -5275,14 +5209,12 @@ class TestSchedulerJob:
with dag_maker("test_dag1", max_active_runs=1):
EmptyOperator(task_id="mytask")
date = DEFAULT_DATE
- dag_version = DagVersion.get_latest_version("test_dag1")
for i in range(30):
dr = dag_maker.create_dagrun(
run_id=f"dagrun_{i}",
run_type=DagRunType.SCHEDULED,
state=State.QUEUED,
logical_date=date,
- dag_version=dag_version,
)
date = dr.logical_date + timedelta(hours=1)
scheduler_job = Job(executor=MockExecutor(do_update=False))
@@ -5322,13 +5254,11 @@ class TestSchedulerJob:
with dag_maker("test_dagrun_states_are_correct_1", max_active_runs=1, start_date=date) as dag:
task1 = EmptyOperator(task_id="dummy_task")
- dag_version = DagVersion.get_latest_version(dag.dag_id)
- dr1_running = dag_maker.create_dagrun(run_id="dr1_run_1", logical_date=date, dag_version=dag_version)
+ dr1_running = dag_maker.create_dagrun(run_id="dr1_run_1", logical_date=date)
data_interval = dag.infer_automated_data_interval(logical_date)
dag_maker.create_dagrun(
run_id="dr1_run_2",
state=State.QUEUED,
- dag_version=dag_version,
logical_date=dag.next_dagrun_info(
last_automated_dagrun=data_interval, restricted=False
).data_interval.start,
@@ -5337,13 +5267,11 @@ class TestSchedulerJob:
date = timezone.datetime(2020, 1, 1)
with dag_maker("test_dagrun_states_are_correct_2", start_date=date) as dag:
EmptyOperator(task_id="dummy_task")
- dag_version = DagVersion.get_latest_version(dag.dag_id)
for i in range(16):
dr = dag_maker.create_dagrun(
run_id=f"dr2_run_{i + 1}",
state=State.RUNNING,
logical_date=date,
- dag_version=dag_version,
)
date = dr.logical_date + timedelta(hours=1)
dr16 = DagRun.find(run_id="dr2_run_16")
@@ -5353,7 +5281,6 @@ class TestSchedulerJob:
run_id=f"dr2_run_{i + 1}",
state=State.QUEUED,
logical_date=date,
- dag_version=dag_version,
)
date = dr.logical_date + timedelta(hours=1)
@@ -5361,13 +5288,11 @@ class TestSchedulerJob:
date = timezone.datetime(2021, 1, 1)
with dag_maker("test_dagrun_states_are_correct_3", start_date=date) as dag:
EmptyOperator(task_id="dummy_task")
- dag_version = DagVersion.get_latest_version(dag.dag_id)
for i in range(16):
dr = dag_maker.create_dagrun(
run_id=f"dr3_run_{i + 1}",
state=State.RUNNING,
logical_date=date,
- dag_version=dag_version,
)
date = dr.logical_date + timedelta(hours=1)
dr16 = DagRun.find(run_id="dr3_run_16")
@@ -5377,7 +5302,6 @@ class TestSchedulerJob:
run_id=f"dr2_run_{i + 1}",
state=State.QUEUED,
logical_date=date,
- dag_version=dag_version,
)
date = dr.logical_date + timedelta(hours=1)
@@ -5647,17 +5571,10 @@ class TestSchedulerJob:
scheduler_job = Job()
self.job_runner = SchedulerJobRunner(job=scheduler_job)
- dag_version = DagVersion.get_latest_version(dag.dag_id)
session = settings.Session()
- dr = dag_maker.create_dagrun(
- run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
- dr = dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
- dag_maker.create_dagrun_after(
- dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED, dag_version=dag_version
- )
+ dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED)
+ dr = dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
+ dag_maker.create_dagrun_after(dr, run_type=DagRunType.SCHEDULED, state=State.QUEUED)
dag.clear()
assert len(DagRun.find(dag_id=dag.dag_id, state=State.QUEUED, session=session)) == 3
diff --git a/airflow-core/tests/unit/models/test_dag.py b/airflow-core/tests/unit/models/test_dag.py
index b90c985cc11..0c7457f587d 100644
--- a/airflow-core/tests/unit/models/test_dag.py
+++ b/airflow-core/tests/unit/models/test_dag.py
@@ -57,7 +57,6 @@ from airflow.models.dag import (
ExecutorLoader,
get_asset_triggered_next_run_info,
)
-from airflow.models.dag_version import DagVersion
from airflow.models.dagrun import DagRun
from airflow.models.serialized_dag import SerializedDagModel
from airflow.models.taskinstance import TaskInstance as TI
@@ -947,7 +946,6 @@ class TestDag:
def test_existing_dag_is_paused_after_limit(self, testing_dag_bundle):
def add_failed_dag_run(dag, id, logical_date):
- dag_v = DagVersion.get_latest_version(dag_id=dag.dag_id)
dr = dag.create_dagrun(
run_type=DagRunType.MANUAL,
run_id="run_id_" + id,
@@ -955,7 +953,6 @@ class TestDag:
state=State.FAILED,
data_interval=(logical_date, logical_date),
run_after=logical_date,
- dag_version=dag_v,
triggered_by=DagRunTriggeredByType.TEST,
session=session,
)
diff --git a/airflow-core/tests/unit/models/test_dagrun.py b/airflow-core/tests/unit/models/test_dagrun.py
index 35f4ec79648..9d76ffabc1c 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -100,7 +100,6 @@ class TestDagRun:
logical_date: datetime.datetime | None = None,
is_backfill: bool = False,
state: DagRunState = DagRunState.RUNNING,
- dag_version: DagVersion | None = None,
session: Session,
):
now = timezone.utcnow()
@@ -123,7 +122,6 @@ class TestDagRun:
run_after=data_interval.end,
start_date=now,
state=state,
- dag_version=dag_version or DagVersion.get_latest_version(dag.dag_id, session=session),
triggered_by=DagRunTriggeredByType.TEST,
session=session,
)
diff --git a/airflow-core/tests/unit/models/test_taskinstance.py b/airflow-core/tests/unit/models/test_taskinstance.py
index acbcb1939fd..349b211dd4f 100644
--- a/airflow-core/tests/unit/models/test_taskinstance.py
+++ b/airflow-core/tests/unit/models/test_taskinstance.py
@@ -4719,25 +4719,25 @@ class TestMappedTaskInstanceReceiveValue:
def test_map_has_dag_version(self, dag_maker, session):
from airflow.models.dag_version import DagVersion
- known_versions = {}
+ known_versions = []
with dag_maker(dag_id="test", session=session) as dag:
@dag.task
def show(value, *, ti):
- known_versions[ti.map_index] = ti.dag_version_id
+ # let's record the dag version ids we observe on the tis
+ known_versions.append(ti.dag_version_id)
show.expand(value=[1, 2, 3])
-
+ # ensure that there is a dag_version record in the db
dag_version = session.merge(DagVersion(dag_id="test", bundle_name="test"))
-
- dag_maker.create_dagrun(dag_version=dag_version)
+ dag_maker.create_dagrun()
task = dag.get_task("show")
for ti in session.scalars(select(TI)):
ti.refresh_from_task(task)
ti.run()
-
- assert known_versions == {0: dag_version.id, 1: dag_version.id, 2: dag_version.id}
+ # verify that we only saw the dag version we created
+ assert known_versions == [dag_version.id] * 3
@pytest.mark.parametrize(
"upstream_return, expected_outputs",
diff --git a/dev/airflow_perf/scheduler_dag_execution_timing.py b/dev/airflow_perf/scheduler_dag_execution_timing.py
index 689e18bbbd3..47eaf92d7c8 100755
--- a/dev/airflow_perf/scheduler_dag_execution_timing.py
+++ b/dev/airflow_perf/scheduler_dag_execution_timing.py
@@ -176,7 +176,6 @@ def create_dag_runs(dag, num_runs, session):
run_after=logical_date,
run_type=DagRunType.MANUAL,
triggered_by=DagRunTriggeredByType.TEST,
- dag_version=None,
state=DagRunState.RUNNING,
start_date=timezone.utcnow(),
session=session,
diff --git a/devel-common/src/tests_common/pytest_plugin.py b/devel-common/src/tests_common/pytest_plugin.py
index 33cb0fe6ea5..3ae9c7dffc9 100644
--- a/devel-common/src/tests_common/pytest_plugin.py
+++ b/devel-common/src/tests_common/pytest_plugin.py
@@ -995,10 +995,8 @@ def dag_maker(request) -> Generator[DagMaker, None, None]:
if AIRFLOW_V_3_0_PLUS:
kwargs.setdefault("triggered_by", DagRunTriggeredByType.TEST)
kwargs["logical_date"] = logical_date
- kwargs.setdefault("dag_version", None)
kwargs.setdefault("run_after", data_interval[-1] if data_interval else timezone.utcnow())
else:
- kwargs.pop("dag_version", None)
kwargs.pop("triggered_by", None)
kwargs["execution_date"] = logical_date
diff --git a/providers/standard/tests/unit/standard/sensors/test_external_task_sensor.py b/providers/standard/tests/unit/standard/sensors/test_external_task_sensor.py
index efa33f5a558..2ecb958f62b 100644
--- a/providers/standard/tests/unit/standard/sensors/test_external_task_sensor.py
+++ b/providers/standard/tests/unit/standard/sensors/test_external_task_sensor.py
@@ -1504,7 +1504,6 @@ def run_tasks(
run_after=logical_date,
run_type=DagRunType.MANUAL,
triggered_by=DagRunTriggeredByType.TEST,
- dag_version=None,
state=DagRunState.RUNNING,
start_date=logical_date,
session=session,