[Python]Support pipe operator as Union (PEP -604) (#24106)

Fixes #21972

sdks/python/apache_beam/typehints/native_type_compatibility.py

@@ -22,6 +22,7 @@
 import collections
 import logging
 import sys
+import types
 import typing
 
 from apache_beam.typehints import typehints
@@ -176,6 +177,14 @@ def convert_to_beam_type(typ):
   Raises:
     ValueError: The type was malformed.
   """
+  # Convert `int | float` to typing.Union[int, float]
+  # pipe operator as Union and types.UnionType are introduced
+  # in Python 3.10.
+  # GH issue: https://github.com/apache/beam/issues/21972
+  if (sys.version_info.major == 3 and
+      sys.version_info.minor >= 10) and (isinstance(typ, types.UnionType)):
+    typ = typing.Union[typ]
+
   if isinstance(typ, typing.TypeVar):
     # This is a special case, as it's not parameterized by types.
     # Also, identity must be preserved through conversion (i.e. the same

sdks/python/apache_beam/typehints/typehints.py

@@ -67,6 +67,8 @@
 
 import copy
 import logging
+import sys
+import types
 import typing
 from collections import abc
 
@@ -384,6 +386,9 @@ def validate_composite_type_param(type_param, error_msg_prefix):
       not isinstance(type_param, tuple(possible_classes)) and
       type_param is not None and
       getattr(type_param, '__module__', None) != 'typing')
+  if sys.version_info.major == 3 and sys.version_info.minor >= 10:
+    if isinstance(type_param, types.UnionType):
+      is_not_type_constraint = False
   is_forbidden_type = (
       isinstance(type_param, type) and type_param in DISALLOWED_PRIMITIVE_TYPES)
 

sdks/python/apache_beam/typehints/typehints_test.py

@@ -1599,6 +1599,16 @@ def expand(self, pcoll: typing.Any) -> typehints.Any:
     self.assertEqual(th.input_types, ((typehints.Any, ), {}))
     self.assertEqual(th.input_types, ((typehints.Any, ), {}))
 
+  def test_pipe_operator_as_union(self):
+    # union types can be written using pipe operator from Python 3.10.
+    # https://peps.python.org/pep-0604/
+    if sys.version_info.major == 3 and sys.version_info.minor >= 10:
+      type_a = int | float  # pylint: disable=unsupported-binary-operation
+      type_b = typing.Union[int, float]
+      self.assertEqual(
+          native_type_compatibility.convert_to_beam_type(type_a),
+          native_type_compatibility.convert_to_beam_type(type_b))
+
 
 if __name__ == '__main__':
   unittest.main()