From cb4093935621722327cd5d9e7e78ab35d5d704c3 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Fri, 27 Sep 2024 11:00:24 -0700
Subject: [PATCH 1/2] Remove DagRun.is_backfill attribute

This attribute is only used in one place and is not very useful.
---
 airflow/models/dagrun.py         | 6 +-----
 tests/jobs/test_scheduler_job.py | 5 ++---
 2 files changed, 3 insertions(+), 8 deletions(-)

diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
index 3ef1c18f152a4..5d53e51763dff 100644
--- a/airflow/models/dagrun.py
+++ b/airflow/models/dagrun.py
@@ -1267,7 +1267,7 @@ def verify_integrity(self, *, session: Session = NEW_SESSION) -> None:
 
         def task_filter(task: Operator) -> bool:
             return task.task_id not in task_ids and (
-                self.is_backfill
+                self.run_type == DagRunType.BACKFILL_JOB
                 or (task.start_date is None or task.start_date <= self.execution_date)
                 and (task.end_date is None or self.execution_date <= task.end_date)
             )
@@ -1538,10 +1538,6 @@ def _revise_map_indexes_if_mapped(self, task: Operator, *, session: Session) ->
             session.flush()
             yield ti
 
-    @property
-    def is_backfill(self) -> bool:
-        return self.run_type == DagRunType.BACKFILL_JOB
-
     @classmethod
     @provide_session
     def get_latest_runs(cls, session: Session = NEW_SESSION) -> list[DagRun]:
diff --git a/tests/jobs/test_scheduler_job.py b/tests/jobs/test_scheduler_job.py
index 52e9dbdeb1a04..78a911153dab2 100644
--- a/tests/jobs/test_scheduler_job.py
+++ b/tests/jobs/test_scheduler_job.py
@@ -601,8 +601,7 @@ def test_execute_task_instances_backfill_tasks_wont_execute(self, dag_maker):
         ti1.state = State.SCHEDULED
         session.merge(ti1)
         session.flush()
-
-        assert dr1.is_backfill
+        assert dr1.run_type == DagRunType.BACKFILL_JOB
 
         self.job_runner._critical_section_enqueue_task_instances(session)
         session.flush()
@@ -3851,7 +3850,7 @@ def test_adopt_or_reset_orphaned_tasks_backfill_dag(self, dag_maker):
         session.merge(dr1)
         session.flush()
 
-        assert dr1.is_backfill
+        assert dr1.run_type == DagRunType.BACKFILL_JOB
         assert 0 == self.job_runner.adopt_or_reset_orphaned_tasks(session=session)
         session.rollback()
 

From ecf75733fbaf146f872f43a7d23b23824645d46f Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Fri, 27 Sep 2024 12:44:25 -0700
Subject: [PATCH 2/2] add newsfragment

---
 newsfragments/42548.significant.rst | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 newsfragments/42548.significant.rst

diff --git a/newsfragments/42548.significant.rst b/newsfragments/42548.significant.rst
new file mode 100644
index 0000000000000..28d6795eebcc6
--- /dev/null
+++ b/newsfragments/42548.significant.rst
@@ -0,0 +1 @@
+Remove is_backfill attribute from DagRun object