Mypy & ruff

Author: TheoS

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

@@ -1248,7 +1248,7 @@ def recalculate(self) -> _UnfinishedStates:
             if dag.has_on_success_callback:
                 last_succeeded_ti: TI | None = max(
                     (ti for ti in tis if ti.state == TaskInstanceState.SUCCESS),
-                    key=lambda ti: ti.end_date,
+                    key=lambda ti: ti.end_date or timezone.make_aware(datetime.min),
                     default=None,
                 )
                 callback = self.produce_dag_callback(
@@ -1281,7 +1281,8 @@ def recalculate(self) -> _UnfinishedStates:
                 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 not (
+                    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)

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

@@ -498,7 +498,7 @@ def on_failure_callable(context):
             middle.trigger_rule = TriggerRule.ONE_FAILED
             middle.set_upstream(up)
             middle.set_downstream(down)
-        
+
         dr = dag_maker.create_dagrun()
 
         ti_up: TI = dr.get_task_instance(task_id=up.task_id, session=session)
@@ -507,7 +507,7 @@ def on_failure_callable(context):
         ti_middle.set_state(state=None, session=session)
         ti_middle.task.trigger_rule = "invalid"
 
-        serialized_dag = dr.get_dag()  
+        serialized_dag = dr.get_dag()
 
         with mock.patch.object(dr, "execute_dag_callbacks") as execute_dag_callbacks:
             _, callback = dr.update_state()