This is an automated email from the ASF dual-hosted git repository.
Lee-W pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 7b6512ceb87 [v3-3-test] Return 422 for empty backfill window and stop
leaving orphan rows (#68883) (#69367)
7b6512ceb87 is described below
commit 7b6512ceb87aeff445ac19a9000c63b658084006
Author: Wei Lee <[email protected]>
AuthorDate: Sat Jul 4 18:23:48 2026 +0800
[v3-3-test] Return 422 for empty backfill window and stop leaving orphan
rows (#68883) (#69367)
---
.../core_api/routes/public/backfills.py | 2 +
.../src/airflow/cli/commands/backfill_command.py | 48 +++++------
airflow-core/src/airflow/models/backfill.py | 31 +++++---
.../core_api/routes/public/test_backfills.py | 58 +++++++++++++-
.../unit/cli/commands/test_backfill_command.py | 61 +++++++++++++-
airflow-core/tests/unit/models/test_backfill.py | 92 +++++++++++++++++++++-
6 files changed, 255 insertions(+), 37 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
index 0dba6086e0a..70c353fb7a2 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
@@ -56,6 +56,7 @@ from airflow.models.backfill import (
InvalidBackfillDateRange,
InvalidBackfillDirection,
InvalidReprocessBehavior,
+ NoBackfillRunsToCreate,
_create_backfill,
_do_dry_run,
)
@@ -276,6 +277,7 @@ def create_backfill(
InvalidBackfillDate,
InvalidBackfillDateRange,
InvalidBackfillConf,
+ NoBackfillRunsToCreate,
) as e:
raise RequestValidationError(str(e))
diff --git a/airflow-core/src/airflow/cli/commands/backfill_command.py
b/airflow-core/src/airflow/cli/commands/backfill_command.py
index d76d435a391..75ba6b03ce4 100644
--- a/airflow-core/src/airflow/cli/commands/backfill_command.py
+++ b/airflow-core/src/airflow/cli/commands/backfill_command.py
@@ -27,7 +27,7 @@ from airflow import settings
from airflow.api_fastapi.common.dagbag import resolve_run_on_latest_version
from airflow.cli.simple_table import AirflowConsole
from airflow.exceptions import AirflowConfigException
-from airflow.models.backfill import ReprocessBehavior, _create_backfill,
_do_dry_run
+from airflow.models.backfill import NoBackfillRunsToCreate, ReprocessBehavior,
_create_backfill, _do_dry_run
from airflow.utils import cli as cli_utils
from airflow.utils.cli import sigint_handler
from airflow.utils.platform import getuser
@@ -50,6 +50,13 @@ def create_backfill(args) -> None:
else:
reprocess_behavior = None
+ dag_run_conf = None
+ if args.dag_run_conf:
+ try:
+ dag_run_conf = json.loads(args.dag_run_conf)
+ except json.JSONDecodeError as e:
+ raise ValueError(f"Invalid JSON in --dag-run-conf: {e}")
+
with create_session() as session:
resolved_run_on_latest = resolve_run_on_latest_version(
args.run_on_latest_version,
@@ -67,7 +74,7 @@ def create_backfill(args) -> None:
to_date=args.to_date,
max_active_runs=args.max_active_runs,
reverse=args.run_backwards,
- dag_run_conf=args.dag_run_conf,
+ dag_run_conf=dag_run_conf,
reprocess_behavior=reprocess_behavior,
run_on_latest_version=resolved_run_on_latest,
)
@@ -79,7 +86,8 @@ def create_backfill(args) -> None:
from_date=args.from_date,
to_date=args.to_date,
reverse=args.run_backwards,
- reprocess_behavior=args.reprocess_behavior,
+ reprocess_behavior=reprocess_behavior or
ReprocessBehavior.NONE,
+ dag_run_conf=dag_run_conf,
session=session,
)
console.print("Runs to be attempted:")
@@ -97,22 +105,18 @@ def create_backfill(args) -> None:
log.warning("Failed to get user name from os: %s, not setting the
triggering user", e)
user = None
- # Parse dag_run_conf if provided
- dag_run_conf = None
- if args.dag_run_conf:
- try:
- dag_run_conf = json.loads(args.dag_run_conf)
- except json.JSONDecodeError as e:
- raise ValueError(f"Invalid JSON in --dag-run-conf: {e}")
-
- _create_backfill(
- dag_id=args.dag_id,
- from_date=args.from_date,
- to_date=args.to_date,
- max_active_runs=args.max_active_runs,
- reverse=args.run_backwards,
- dag_run_conf=dag_run_conf,
- triggering_user_name=user,
- reprocess_behavior=reprocess_behavior,
- run_on_latest_version=resolved_run_on_latest,
- )
+ try:
+ _create_backfill(
+ dag_id=args.dag_id,
+ from_date=args.from_date,
+ to_date=args.to_date,
+ max_active_runs=args.max_active_runs,
+ reverse=args.run_backwards,
+ dag_run_conf=dag_run_conf,
+ triggering_user_name=user,
+ reprocess_behavior=reprocess_behavior,
+ run_on_latest_version=resolved_run_on_latest,
+ )
+ except NoBackfillRunsToCreate as e:
+ console.print(f"[yellow]Warning:[/yellow] {e}")
+ raise SystemExit(1)
diff --git a/airflow-core/src/airflow/models/backfill.py
b/airflow-core/src/airflow/models/backfill.py
index 52019fa5d98..1f507de94bb 100644
--- a/airflow-core/src/airflow/models/backfill.py
+++ b/airflow-core/src/airflow/models/backfill.py
@@ -123,6 +123,17 @@ class InvalidBackfillConf(AirflowException):
"""
+class NoBackfillRunsToCreate(ValueError):
+ """
+ Raised when a backfill request yields no Dag runs for the given date range.
+
+ This happens when the from/to date range falls entirely outside the Dag's
+ scheduled intervals (e.g. the range predates the first partition boundary).
+
+ :meta private:
+ """
+
+
class UnknownActiveBackfills(AirflowException):
"""
Raised when the quantity of active backfills cannot be determined.
@@ -665,6 +676,17 @@ def _create_backfill(
dag_run_conf,
)
+ dagrun_info_list = _get_info_list(
+ from_date=from_date,
+ to_date=to_date,
+ reverse=reverse,
+ dag=dag,
+ )
+ if not dagrun_info_list:
+ raise NoBackfillRunsToCreate(
+ f"No runs to create for Dag {dag_id} in the range
[{from_date}, {to_date}]"
+ )
+
br = Backfill(
dag_id=dag_id,
from_date=from_date,
@@ -680,15 +702,6 @@ def _create_backfill(
session.scalars(select(DagModel).where(DagModel.dag_id ==
dag_id)).one()
- dagrun_info_list = _get_info_list(
- from_date=from_date,
- to_date=to_date,
- reverse=reverse,
- dag=dag,
- )
- if not dagrun_info_list:
- raise RuntimeError(f"No runs to create for Dag {dag_id}")
-
first_info = dagrun_info_list[0]
if first_info.partition_key:
_create_runs_partitioned(
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
index 42b48fc0c49..2380f73b9bf 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
@@ -27,7 +27,13 @@ from sqlalchemy import and_, func, select
from airflow._shared.timezones import timezone
from airflow.dag_processing.dagbag import DagBag
from airflow.models import DagModel, DagRun
-from airflow.models.backfill import Backfill, BackfillDagRun,
ReprocessBehavior, _create_backfill
+from airflow.models.backfill import (
+ Backfill,
+ BackfillDagRun,
+ NoBackfillRunsToCreate,
+ ReprocessBehavior,
+ _create_backfill,
+)
from airflow.models.dag import DAG
from airflow.models.dagbundle import DagBundleModel
from airflow.providers.standard.operators.empty import EmptyOperator
@@ -965,6 +971,56 @@ class TestCreateBackfillPartitioned(TestBackfillEndpoint):
response = test_client.post(url=url, json=data)
assert response.status_code == 422
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.backfills._create_backfill",
autospec=True)
+ def test_empty_window_create_returns_422(self, mock_create, session,
dag_maker, test_client):
+ """POST /backfills with an empty window raises NoBackfillRunsToCreate
→ 422."""
+ mock_create.side_effect = NoBackfillRunsToCreate(
+ "No runs to create for Dag TEST_PARTITIONED_DAG in the range [...]"
+ )
+ with dag_maker(
+ session=session,
+ dag_id="TEST_PARTITIONED_DAG",
+ schedule=CronPartitionTimetable("0 0 * * *",
timezone="Asia/Taipei"),
+ ):
+ EmptyOperator(task_id="mytask")
+ session.commit()
+
+ data = {
+ "dag_id": "TEST_PARTITIONED_DAG",
+ "from_date": "2026-02-18T00:00:00+00:00",
+ "to_date": "2026-02-18T00:00:00+00:00",
+ "max_active_runs": 5,
+ "run_backwards": False,
+ }
+ response = test_client.post(url="/backfills", json=data)
+ assert response.status_code == 422
+
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.backfills._do_dry_run",
autospec=True)
+ def test_empty_window_dry_run_returns_200_with_zero_entries(
+ self, mock_dry_run, session, dag_maker, test_client
+ ):
+ """POST /backfills/dry_run with an empty window returns 200 with
total_entries == 0."""
+ mock_dry_run.return_value = iter([])
+ with dag_maker(
+ session=session,
+ dag_id="TEST_PARTITIONED_DAG",
+ schedule=CronPartitionTimetable("0 0 * * *",
timezone="Asia/Taipei"),
+ ):
+ EmptyOperator(task_id="mytask")
+ session.commit()
+
+ data = {
+ "dag_id": "TEST_PARTITIONED_DAG",
+ "from_date": "2026-02-18T00:00:00+00:00",
+ "to_date": "2026-02-18T00:00:00+00:00",
+ "max_active_runs": 5,
+ "run_backwards": False,
+ }
+ response = test_client.post(url="/backfills/dry_run", json=data)
+ assert response.status_code == 200
+ assert response.json()["total_entries"] == 0
+ assert response.json()["backfills"] == []
+
class TestCancelBackfill(TestBackfillEndpoint):
def test_cancel_backfill(self, session, test_client):
diff --git a/airflow-core/tests/unit/cli/commands/test_backfill_command.py
b/airflow-core/tests/unit/cli/commands/test_backfill_command.py
index c897fa20443..5d5743349f7 100644
--- a/airflow-core/tests/unit/cli/commands/test_backfill_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_backfill_command.py
@@ -164,7 +164,8 @@ class TestCliBackfill:
from_date=DEFAULT_DATE.replace(tzinfo=timezone.utc),
to_date=DEFAULT_DATE.replace(tzinfo=timezone.utc),
reverse=reverse,
- reprocess_behavior="none",
+ reprocess_behavior=ReprocessBehavior.NONE,
+ dag_run_conf=None,
session=mock.ANY,
)
@@ -240,9 +241,10 @@ class TestCliBackfill:
with pytest.raises(SystemExit):
self.parser.parse_args(args)
+ @mock.patch("airflow.cli.commands.backfill_command.getuser",
return_value="test_user")
@mock.patch("airflow.cli.commands.backfill_command._create_backfill")
- def test_backfill_with_empty_dag_run_conf(self, mock_create):
- """Test that empty dag_run_conf is properly parsed."""
+ def test_backfill_with_empty_dag_run_conf(self, mock_create, mock_getuser):
+ """Test that empty dag_run_conf ({}) is parsed as an empty dict, not
None."""
args = [
"backfill",
"create",
@@ -265,6 +267,57 @@ class TestCliBackfill:
reverse=False,
dag_run_conf={},
reprocess_behavior=None,
- triggering_user_name="root",
+ triggering_user_name="test_user",
run_on_latest_version=True,
)
+
+ @mock.patch("airflow.cli.commands.backfill_command._do_dry_run",
autospec=True)
+ def test_backfill_dry_run_passes_dag_run_conf(self, mock_dry_run):
+ """dry-run path forwards parsed dag_run_conf dict (not raw string) to
_do_dry_run."""
+ mock_dry_run.return_value = iter([])
+ args = [
+ "backfill",
+ "create",
+ "--dag-id",
+ "example_bash_operator",
+ "--from-date",
+ DEFAULT_DATE.isoformat(),
+ "--to-date",
+ DEFAULT_DATE.isoformat(),
+ "--dry-run",
+ "--reprocess-behavior",
+ "failed",
+ "--dag-run-conf",
+ '{"key": "val"}',
+ ]
+
airflow.cli.commands.backfill_command.create_backfill(self.parser.parse_args(args))
+
+ mock_dry_run.assert_called_once_with(
+ dag_id="example_bash_operator",
+ from_date=DEFAULT_DATE.replace(tzinfo=timezone.utc),
+ to_date=DEFAULT_DATE.replace(tzinfo=timezone.utc),
+ reverse=False,
+ reprocess_behavior=ReprocessBehavior.FAILED,
+ dag_run_conf={"key": "val"},
+ session=mock.ANY,
+ )
+
+ @mock.patch("airflow.cli.commands.backfill_command._create_backfill",
autospec=True)
+ def test_backfill_empty_window_shows_friendly_message(self, mock_create):
+ """Empty-window backfill exits with code 1 and prints a message, no
traceback."""
+ from airflow.models.backfill import NoBackfillRunsToCreate
+
+ mock_create.side_effect = NoBackfillRunsToCreate("No runs to create
for Dag example_bash_operator")
+ args = [
+ "backfill",
+ "create",
+ "--dag-id",
+ "example_bash_operator",
+ "--from-date",
+ DEFAULT_DATE.isoformat(),
+ "--to-date",
+ DEFAULT_DATE.isoformat(),
+ ]
+ with pytest.raises(SystemExit) as exc_info:
+
airflow.cli.commands.backfill_command.create_backfill(self.parser.parse_args(args))
+ assert exc_info.value.code == 1
diff --git a/airflow-core/tests/unit/models/test_backfill.py
b/airflow-core/tests/unit/models/test_backfill.py
index 69a4ebd5dc2..ea621be84d6 100644
--- a/airflow-core/tests/unit/models/test_backfill.py
+++ b/airflow-core/tests/unit/models/test_backfill.py
@@ -20,10 +20,11 @@ from __future__ import annotations
from contextlib import nullcontext
from datetime import datetime, timedelta
from typing import TYPE_CHECKING
+from unittest import mock
import pendulum
import pytest
-from sqlalchemy import select
+from sqlalchemy import func, select
from airflow._shared.timezones import timezone
from airflow.models import DagModel, DagRun, TaskInstance
@@ -37,6 +38,7 @@ from airflow.models.backfill import (
InvalidBackfillDateRange,
InvalidBackfillDirection,
InvalidReprocessBehavior,
+ NoBackfillRunsToCreate,
ReprocessBehavior,
_create_backfill,
_do_dry_run,
@@ -1384,6 +1386,94 @@ def
test_partitioned_backfill_reprocess_failed(dag_maker, session):
assert bdr.partition_key == info.partition_key
[email protected]("airflow.models.backfill._get_info_list", autospec=True,
return_value=[])
+def
test_create_backfill_empty_window_raises_no_runs_to_create(mock_get_info_list,
dag_maker, session):
+ """_create_backfill raises NoBackfillRunsToCreate when _get_info_list
returns an empty list."""
+ with dag_maker(schedule=CronPartitionTimetable("0 0 * * *",
timezone="Asia/Taipei")) as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+ session.commit()
+
+ with pytest.raises(NoBackfillRunsToCreate, match=dag.dag_id):
+ _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2026-02-18"),
+ to_date=pendulum.parse("2026-02-18"),
+ max_active_runs=2,
+ reverse=False,
+ triggering_user_name="pytest",
+ dag_run_conf={},
+ )
+
+
[email protected]("dag_run_conf", [None, {}])
[email protected]("airflow.models.backfill._get_info_list", autospec=True,
return_value=[])
+def test_do_dry_run_empty_window_returns_empty_iterable(mock_get_info_list,
dag_run_conf, dag_maker, session):
+ """_do_dry_run on an empty window yields nothing (does not raise), with or
without dag_run_conf."""
+ with dag_maker(schedule=CronPartitionTimetable("0 0 * * *",
timezone="Asia/Taipei")) as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+ session.commit()
+
+ infos = list(
+ _do_dry_run(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2026-02-18"),
+ to_date=pendulum.parse("2026-02-18"),
+ reverse=False,
+ reprocess_behavior=ReprocessBehavior.NONE,
+ dag_run_conf=dag_run_conf,
+ session=session,
+ )
+ )
+ assert infos == []
+
+
+def test_create_backfill_real_empty_window_no_orphan(dag_maker, session):
+ """_create_backfill with a real empty window raises NoBackfillRunsToCreate
without leaving an orphan.
+
+ Uses a weekly-Monday timetable; 2026-02-18 is a Wednesday so
_get_info_list returns [] for real.
+ After the raise, no incomplete Backfill row must exist — proving #1
(orphan fix) holds.
+ A second call for the same dag_id must succeed (not be blocked by
AlreadyRunningBackfill).
+ """
+ with dag_maker(schedule=CronPartitionTimetable("0 0 * * 1",
timezone="UTC")) as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+ session.commit()
+
+ wednesday = pendulum.datetime(2026, 2, 18, tz="UTC") # not a Monday
+
+ with pytest.raises(NoBackfillRunsToCreate, match=dag.dag_id):
+ _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=wednesday,
+ to_date=wednesday,
+ max_active_runs=2,
+ reverse=False,
+ triggering_user_name="pytest",
+ dag_run_conf=None,
+ )
+
+ # No orphan: zero incomplete Backfill rows for this dag
+ orphan_count = session.scalar(
+ select(func.count()).where(
+ Backfill.dag_id == dag.dag_id,
+ Backfill.completed_at.is_(None),
+ )
+ )
+ assert orphan_count == 0, "An orphan Backfill row was left behind after
NoBackfillRunsToCreate"
+
+ # A valid (Monday) window must not be blocked by AlreadyRunningBackfill
+ monday = pendulum.datetime(2026, 2, 23, tz="UTC") # a Monday
+ br = _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=monday,
+ to_date=monday,
+ max_active_runs=2,
+ reverse=False,
+ triggering_user_name="pytest",
+ dag_run_conf=None,
+ )
+ assert br is not None
+
+
def test_handle_clear_run_preserves_partition_key(dag_maker, session):
"""BackfillDagRun created via the clear/reprocess path carries
partition_key from info."""
