style: fix provider mypy error

Author: Lee-W

providers/atlassian/jira/src/airflow/providers/atlassian/jira/sensors/jira.py

@@ -52,7 +52,7 @@ def __init__(
         if result_processor is not None:
             self.result_processor = result_processor
         self.method_name = method_name
-        self.method_params = method_params
+        self.method_params = method_params or {}
 
     def poke(self, context: Context) -> Any:
         hook = JiraHook(jira_conn_id=self.jira_conn_id)

providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py

@@ -55,7 +55,7 @@ class EdgeExecutor(BaseExecutor):
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.last_reported_state: dict[TaskInstanceKey, TaskInstanceState] = {}
+        self.last_reported_state: dict[TaskInstanceKey, TaskInstanceState | str] = {}
 
         # Check if self has the ExecutorConf set on the self.conf attribute with all required methods.
         # In Airflow 2.x, ExecutorConf exists but lacks methods like getint, getboolean, getsection, etc.

providers/github/src/airflow/providers/github/sensors/github.py

@@ -54,7 +54,7 @@ def __init__(
         if result_processor is not None:
             self.result_processor = result_processor
         self.method_name = method_name
-        self.method_params = method_params
+        self.method_params = method_params or {}
 
     def poke(self, context: Context) -> bool:
         hook = GithubHook(github_conn_id=self.github_conn_id)

providers/google/src/airflow/providers/google/cloud/triggers/cloud_composer.py

@@ -22,7 +22,7 @@
 import json
 from collections.abc import Collection, Iterable, Sequence
 from datetime import datetime
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from dateutil import parser
 from google.api_core.exceptions import NotFound
@@ -32,6 +32,9 @@
 from airflow.providers.google.cloud.hooks.cloud_composer import CloudComposerAsyncHook
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 
+if TYPE_CHECKING:
+    from airflow.utils.state import TerminalTIState
+
 
 class CloudComposerExecutionTrigger(BaseTrigger):
     """The trigger handles the async communication with the Google Cloud Composer."""
@@ -183,7 +186,7 @@ def __init__(
         composer_dag_id: str,
         start_date: datetime,
         end_date: datetime,
-        allowed_states: list[str],
+        allowed_states: list[str] | list[TerminalTIState],
         composer_dag_run_id: str | None = None,
         gcp_conn_id: str = "google_cloud_default",
         impersonation_chain: str | Sequence[str] | None = None,
@@ -358,7 +361,7 @@ def __init__(
         environment_id: str,
         start_date: datetime,
         end_date: datetime,
-        allowed_states: list[str],
+        allowed_states: list[str] | list[TerminalTIState],
         skipped_states: list[str],
         failed_states: list[str],
         composer_external_dag_id: str,

providers/standard/src/airflow/providers/standard/sensors/external_task.py

@@ -53,6 +53,7 @@
 
 if not AIRFLOW_V_3_0_PLUS:
     from airflow.utils.session import NEW_SESSION, provide_session
+    from airflow.utils.state import TerminalTIState
 
 if AIRFLOW_V_3_2_PLUS:
     from airflow.dag_processing.dagbag import DagBag
@@ -321,7 +322,7 @@ def _poke_af3(self, context: Context, dttm_filter: Sequence[datetime.datetime])
         self._has_checked_existence = True
         ti = context["ti"]
 
-        def _get_count(states: list[str]) -> int:
+        def _get_count(states: list[str] | list[TerminalTIState]) -> int:
             if self.external_task_ids:
                 return ti.get_ti_count(
                     dag_id=self.external_dag_id,

task-sdk/src/airflow/sdk/types.py

@@ -21,7 +21,7 @@
 from collections.abc import Iterable
 from typing import TYPE_CHECKING, Any, NamedTuple, Protocol, TypeAlias
 
-from airflow.sdk.api.datamodels._generated import WeightRule
+from airflow.sdk.api.datamodels._generated import TerminalTIState, WeightRule
 from airflow.sdk.bases.xcom import BaseXCom
 from airflow.sdk.definitions._internal.types import NOTSET, ArgNotSet
 
@@ -160,7 +160,7 @@ def get_ti_count(
         task_group_id: str | None = None,
         logical_dates: list[AwareDatetime] | None = None,
         run_ids: list[str] | None = None,
-        states: list[str] | None = None,
+        states: list[str] | list[TerminalTIState] | None = None,
     ) -> int: ...
 
     @staticmethod