Fix check on subclass for typing.Union in _infer_multiple_outputs…

… for Python 3.10+ (#36728)

* Fix check on subclass for `typing.Union` in `_infer_multiple_outputs` for Python 3.10+

* Limit PEP 604 test by Python 3.10

(cherry picked from commit f1d8297)

airflow/decorators/base.py

@@ -350,7 +350,7 @@ def fake():
         except TypeError:  # Can't evaluate return type.
             return False
         ttype = getattr(return_type, "__origin__", return_type)
-        return issubclass(ttype, Mapping)
+        return isinstance(ttype, type) and issubclass(ttype, Mapping)
 
     def __attrs_post_init__(self):
         if "self" in self.function_signature.parameters:

tests/decorators/test_python.py

@@ -97,7 +97,6 @@ def identity_dict_with_decorator_call(x: int, y: int) -> resolve(annotation):
 
         assert identity_dict_with_decorator_call(5, 5).operator.multiple_outputs is True
 
-    @pytest.mark.skipif(sys.version_info < (3, 8), reason="PEP 589 is implemented in Python 3.8")
     def test_infer_multiple_outputs_typed_dict(self):
         from typing import TypedDict
 
@@ -110,6 +109,27 @@ def t1() -> TypeDictClass:
 
         assert t1().operator.multiple_outputs is True
 
+    # We do not enable `from __future__ import annotations` for particular this test module,
+    # that mean `str | None` annotation would raise TypeError in Python 3.9 and below
+    @pytest.mark.skipif(sys.version_info < (3, 10), reason="PEP 604 is implemented in Python 3.10")
+    def test_infer_multiple_outputs_pep_604_union_type(self):
+        @task_decorator
+        def t1() -> str | None:
+            # Before PEP 604 which are implemented in Python 3.10 `str | None`
+            # returns `types.UnionType` which are class and could be check in `issubclass()`.
+            # However in Python 3.10+ this construction returns object `typing.Union`
+            # which can not be used in `issubclass()`
+            return "foo"
+
+        assert t1().operator.multiple_outputs is False
+
+    def test_infer_multiple_outputs_union_type(self):
+        @task_decorator
+        def t1() -> Union[str, None]:
+            return "foo"
+
+        assert t1().operator.multiple_outputs is False
+
     def test_infer_multiple_outputs_forward_annotation(self):
         if TYPE_CHECKING: