Format

Author: TheoS

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -2386,9 +2386,9 @@ def _schedule_dag_run(
                 .where(TI.state.in_(State.unfinished) | (TI.state.is_(None)))
             ).all()
             last_unfinished_ti = max(
-                    unfinished_task_instances,
-                    key=lambda ti: ti.start_date or timezone.make_aware(datetime.min),
-                    default=None,
+                unfinished_task_instances,
+                key=lambda ti: ti.start_date or timezone.make_aware(datetime.min),
+                default=None,
             )
             for task_instance in unfinished_task_instances:
                 task_instance.state = TaskInstanceState.SKIPPED
@@ -2417,12 +2417,12 @@ def _schedule_dag_run(
                 return None
             dag_run = dag_run_reloaded
             callback_to_execute = dag_run.produce_dag_callback(
-                    dag=dag,
-                    success=False,
-                    relevant_ti=last_unfinished_ti,
-                    reason="timed_out",
-                    execute=False,
-                )
+                dag=dag,
+                success=False,
+                relevant_ti=last_unfinished_ti,
+                reason="timed_out",
+                execute=False,
+            )
 
             dag_run.notify_dagrun_state_changed(msg="timed_out")
             if dag_run.end_date and dag_run.start_date:

airflow-core/src/airflow/models/dagrun.py

@@ -1280,12 +1280,15 @@ def recalculate(self) -> _UnfinishedStates:
             if dag.has_on_failure_callback:
                 unfinished_non_schedulable = (ti for ti in unfinished.tis if ti not in set(schedulable_tis))
                 finished_task_ids = {ti.task_id for ti in finished_tis}
-                blocking_ti = next(iter(
-                    ti for ti in unfinished_non_schedulable
-                    if ti.task and not (
-                        ti.task.get_direct_relative_ids(upstream=True).isdisjoint(finished_task_ids)
-                    )
-                ), None)
+                blocking_ti = next(
+                    iter(
+                        ti
+                        for ti in unfinished_non_schedulable
+                        if ti.task
+                        and not (ti.task.get_direct_relative_ids(upstream=True).isdisjoint(finished_task_ids))
+                    ),
+                    None,
+                )
                 callback = self.produce_dag_callback(
                     dag=dag,
                     success=False,

airflow-core/tests/unit/models/test_dagrun.py

@@ -490,7 +490,7 @@ def on_failure_callable(context):
             schedule=datetime.timedelta(days=1),
             start_date=datetime.datetime(2017, 1, 1),
             on_failure_callback=on_failure_callable,
-        ) as dag:
+        ):
             up = EmptyOperator(task_id="upstream")
             middle = EmptyOperator(task_id="wrong")
             down = EmptyOperator(task_id="downstream")