Update routing of types into convert_to_beam_type in normalize() (#34103)

* Update routing of types into convert_to_beam_type in normalize()

* fix typo

* add some quick test cases

Author: jrmccluskey

sdks/python/apache_beam/typehints/typehints.py

@@ -1309,14 +1309,14 @@ def normalize(x, none_as_type=False):
   # Avoid circular imports
   from apache_beam.typehints import native_type_compatibility
 
-  if isinstance(x, types.GenericAlias):
-    x = native_type_compatibility.convert_builtin_to_typing(x)
-
   if none_as_type and x is None:
     return type(None)
+  # Convert bare builtin types to correct type hints directly
   elif x in _KNOWN_PRIMITIVE_TYPES:
     return _KNOWN_PRIMITIVE_TYPES[x]
-  elif getattr(x, '__module__', None) == 'typing':
+  elif getattr(x, '__module__',
+               None) in ('typing', 'collections.abc') or getattr(
+                   x, '__origin__', None) in _KNOWN_PRIMITIVE_TYPES:
     beam_type = native_type_compatibility.convert_to_beam_type(x)
     if beam_type != x:
       # We were able to do the conversion.

sdks/python/apache_beam/typehints/typehints_test.py

@@ -898,6 +898,12 @@ def test_getitem_invalid_composite_type_param(self):
         'an instance of int.',
         e.exception.args[0])
 
+  def test_normalize(self):
+    expected_beam_type = typehints.Sequence[int]
+    converted_beam_type = typehints.normalize(
+        collections.abc.Sequence[int], False)
+    self.assertEqual(converted_beam_type, expected_beam_type)
+
 
 class IterableHintTestCase(TypeHintTestCase):
   def test_getitem_invalid_composite_type_param(self):
@@ -983,6 +989,12 @@ def test_type_check_violation_valid_composite_type(self):
     l = ([[1, 2], [3, 4, 5]])
     self.assertIsNone(hint.type_check(l))
 
+  def test_normalize(self):
+    expected_beam_type = typehints.Iterable[str]
+    converted_beam_type = typehints.normalize(
+        collections.abc.Iterable[str], False)
+    self.assertEqual(converted_beam_type, expected_beam_type)
+
 
 class TestGeneratorWrapper(TypeHintTestCase):
   def test_functions_as_regular_generator(self):