[v3-0-test] Bump mypy to 1.16.1 for airflow-core and fix airflow-core mypy errors (apache#53004)

* Bump mypy to 1.16.1 and fix airflow-core mypy errors

* Bump mypy to 1.16.1 and fix airflow-core mypy errors

* remove mypy dep from additional_dependencies

* resolve comments
(cherry picked from commit bd38225)

Co-authored-by: GPK <[email protected]>

Author: gopidesupavan

airflow-core/src/airflow/api/common/trigger_dag.py

@@ -67,6 +67,7 @@ def _trigger_dag(
         raise DagNotFound(f"Dag id {dag_id} not found")
 
     run_after = run_after or timezone.coerce_datetime(timezone.utcnow())
+    coerced_logical_date: datetime | None = None
     if logical_date:
         if not timezone.is_localized(logical_date):
             raise ValueError("The logical date should be localized")
@@ -84,7 +85,6 @@ def _trigger_dag(
         coerced_logical_date = timezone.coerce_datetime(logical_date)
         data_interval = dag.timetable.infer_manual_data_interval(run_after=run_after)
     else:
-        coerced_logical_date = None
         data_interval = None
 
     run_id = run_id or DagRun.generate_run_id(

airflow-core/src/airflow/api_fastapi/core_api/routes/ui/structure.py

@@ -141,7 +141,7 @@ def structure_data(
                     }
                 )
 
-        if asset_expression := serialized_dag.dag_model.asset_expression:
+        if (asset_expression := serialized_dag.dag_model.asset_expression) and entry_node_ref:
             upstream_asset_nodes, upstream_asset_edges = get_upstream_assets(
                 asset_expression, entry_node_ref["id"]
             )

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

@@ -301,7 +301,7 @@ def __init__(self, **kwargs):
         def dag(self) -> SchedulerDAG:  # type: ignore[override]
             return super().dag  # type: ignore[return-value]
 
-        @dag.setter
+        @dag.setter  # type: ignore[override]
         def dag(self, val: SchedulerDAG):
             # For type checking only
             ...

airflow-core/src/airflow/serialization/serde.py

@@ -273,7 +273,12 @@ def deserialize(o: T | None, full=True, type_hint: Any = None) -> object:
                 class_version,
             )
 
-        return cls(**deserialize(value))
+        deserialize_value = deserialize(value)
+        if not isinstance(deserialize_value, dict):
+            raise TypeError(
+                f"deserialized value for {classname} is not a dict, got {type(deserialize_value)}"
+            )
+        return cls(**deserialize_value)  # type: ignore[operator]
 
     # no deserializer available
     raise TypeError(f"No deserializer found for {classname}")

airflow-core/src/airflow/serialization/serialized_objects.py

@@ -2164,7 +2164,7 @@ def get_task_assets(
                 ports_getter = self._get_mapped_operator_ports
             else:
                 ports_getter = self._get_base_operator_ports
-            directions = ("inlets",) if inlets else ()
+            directions: tuple[str, ...] = ("inlets",) if inlets else ()
             if outlets:
                 directions += ("outlets",)
             for direction in directions:

airflow-core/tests/unit/dags/test_task_view_type_check.py

@@ -42,7 +42,7 @@ def __call__(self):
         """A __call__ method"""
 
 
-def a_function(_, __):
+def a_function(arg_x, __):
     """A function with two args"""