(airflow) branch main updated: Fix mypy type errors in trigger.py (#57457)

[vi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

vincbeck 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 82153db46a5 Fix mypy type errors in trigger.py (#57457)
82153db46a5 is described below

commit 82153db46a5d0ddb2547cbc34c736cedec1e65e7
Author: LI,JHE-CHEN <10...@users.noreply.github.com>
AuthorDate: Thu Oct 30 10:46:46 2025 -0400

    Fix mypy type errors in trigger.py (#57457)
---
 airflow-core/src/airflow/models/trigger.py | 21 ++++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git a/airflow-core/src/airflow/models/trigger.py b/airflow-core/src/airflow/models/trigger.py
index a0d93210870..cd337c69139 100644
--- a/airflow-core/src/airflow/models/trigger.py
+++ b/airflow-core/src/airflow/models/trigger.py
@@ -41,6 +41,7 @@ from airflow.utils.sqlalchemy import UtcDateTime, get_dialect_name, mapped_colum
 from airflow.utils.state import TaskInstanceState
 
 if TYPE_CHECKING:
+    from sqlalchemy import Row
     from sqlalchemy.sql import Select
 
     from airflow.triggers.base import BaseTrigger, TriggerEvent
@@ -233,10 +234,14 @@ class Trigger(Base):
         )
         if get_dialect_name(session) == "mysql":
             # MySQL doesn't support DELETE with JOIN, so we need to do it in two steps
-            ids = session.scalars(ids).all()
-        session.execute(
-            delete(Trigger).where(Trigger.id.in_(ids)).execution_options(synchronize_session=False)
-        )
+            ids_list = list(session.scalars(ids).all())
+            session.execute(
+                delete(Trigger).where(Trigger.id.in_(ids_list)).execution_options(synchronize_session=False)
+            )
+        else:
+            session.execute(
+                delete(Trigger).where(Trigger.id.in_(ids)).execution_options(synchronize_session=False)
+            )
 
     @classmethod
     @provide_session
@@ -308,7 +313,7 @@ class Trigger(Base):
     @provide_session
     def ids_for_triggerer(cls, triggerer_id, session: Session = NEW_SESSION) -> list[int]:
         """Retrieve a list of trigger ids."""
-        return session.scalars(select(cls.id).where(cls.triggerer_id == triggerer_id)).all()
+        return list(session.scalars(select(cls.id).where(cls.triggerer_id == triggerer_id)).all())
 
     @classmethod
     @provide_session
@@ -349,7 +354,7 @@ class Trigger(Base):
         if trigger_ids_query:
             session.execute(
                 update(cls)
-                .where(cls.id.in_([i.id for i in trigger_ids_query]))
+                .where(cls.id.in_([i[0] for i in trigger_ids_query]))
                 .values(triggerer_id=triggerer_id)
                 .execution_options(synchronize_session=False)
             )
@@ -365,7 +370,7 @@ class Trigger(Base):
         :param alive_triggerer_ids: The alive triggerer ids as a list or a select query.
         :param session: The database session.
         """
-        result: list[int] = []
+        result: list[Row[Any]] = []
 
         # Add triggers associated to deadlines first, then tasks, then assets
         # It prioritizes deadline triggers, then DAGs over event driven scheduling which is fair
@@ -447,6 +452,8 @@ def _(event: BaseTaskEndEvent, *, task_instance: TaskInstance, session: Session)
     def _submit_callback_if_necessary() -> None:
         """Submit a callback request if the task state is SUCCESS or FAILED."""
         if event.task_instance_state in (TaskInstanceState.SUCCESS, TaskInstanceState.FAILED):
+            if task_instance.dag_model.relative_fileloc is None:
+                raise RuntimeError("relative_fileloc should not be None for a finished task")
             request = TaskCallbackRequest(
                 filepath=task_instance.dag_model.relative_fileloc,
                 ti=task_instance,