Update typehint code to allow primitive composite types in python 3.9+

sdks/python/apache_beam/typehints/native_type_compatibility_test.py

@@ -114,16 +114,20 @@ def test_convert_to_beam_type(self):
 
   def test_convert_to_beam_type_with_builtin_types(self):
     if sys.version_info >= (3, 9):
-      test_cases = [
-          ('builtin dict', dict[str, int], typehints.Dict[str, int]),
-          ('builtin list', list[str], typehints.List[str]),
-          ('builtin tuple', tuple[str], typehints.Tuple[str]),
-          ('builtin set', set[str], typehints.Set[str]),
-          (
-              'nested builtin',
-              dict[str, list[tuple[float]]],
-              typehints.Dict[str, typehints.List[typehints.Tuple[float]]]),
-      ]
+      test_cases = [('builtin dict', dict[str, int], typehints.Dict[str, int]),
+                    ('builtin list', list[str], typehints.List[str]),
+                    ('builtin tuple', tuple[str], typehints.Tuple[str]),
+                    ('builtin set', set[str], typehints.Set[str]),
+                    (
+                        'nested builtin',
+                        dict[str, list[tuple[float]]],
+                        typehints.Dict[str,
+                                       typehints.List[typehints.Tuple[float]]]),
+                    (
+                        'builtin nested tuple',
+                        tuple[str, list],
+                        typehints.Tuple[str, typehints.List[typehints.Any]],
+                    )]
 
       for test_case in test_cases:
         description = test_case[0]

sdks/python/apache_beam/typehints/typehints.py

@@ -389,10 +389,14 @@ def validate_composite_type_param(type_param, error_msg_prefix):
   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)
-
-  if is_not_type_constraint or is_forbidden_type:
+  # Pre-Python 3.9 compositve type-hinting with built-in types was not
+  # supported, the typing module equivalents should be used instead.
+  if sys.version_info.major == 3 and sys.version_info.minor < 9:
+    is_not_type_constraint = is_not_type_constraint or (
+        isinstance(type_param, type) and
+        type_param in DISALLOWED_PRIMITIVE_TYPES)
+
+  if is_not_type_constraint:
     raise TypeError(
         '%s must be a non-sequence, a type, or a TypeConstraint. %s'
         ' is an instance of %s.' %

sdks/python/apache_beam/typehints/typehints_test.py

@@ -379,9 +379,15 @@ def test_getitem_params_must_be_type_or_constraint(self):
       typehints.Tuple[5, [1, 3]]
     self.assertTrue(e.exception.args[0].startswith(expected_error_prefix))
 
-    with self.assertRaises(TypeError) as e:
-      typehints.Tuple[list, dict]
-    self.assertTrue(e.exception.args[0].startswith(expected_error_prefix))
+    if sys.version_info < (3, 9):
+      with self.assertRaises(TypeError) as e:
+        typehints.Tuple[list, dict]
+      self.assertTrue(e.exception.args[0].startswith(expected_error_prefix))
+    else:
+      try:
+        typehints.Tuple[list, dict]
+      except TypeError:
+        self.fail("built-in composite raised TypeError unexpectedly")
 
   def test_compatibility_arbitrary_length(self):
     self.assertNotCompatible(
@@ -665,8 +671,14 @@ def test_getitem_param_must_have_length_2(self):
         e.exception.args[0])
 
   def test_key_type_must_be_valid_composite_param(self):
-    with self.assertRaises(TypeError):
-      typehints.Dict[list, int]
+    if sys.version_info < (3, 9):
+      with self.assertRaises(TypeError):
+        typehints.Dict[list, int]
+    else:
+      try:
+        typehints.Tuple[list, int]
+      except TypeError:
+        self.fail("built-in composite raised TypeError unexpectedly")
 
   def test_value_type_must_be_valid_composite_param(self):
     with self.assertRaises(TypeError):
@@ -765,13 +777,19 @@ def test_builtin_and_type_compatibility(self):
 class BaseSetHintTest:
   class CommonTests(TypeHintTestCase):
     def test_getitem_invalid_composite_type_param(self):
-      with self.assertRaises(TypeError) as e:
-        self.beam_type[list]
-      self.assertEqual(
-          "Parameter to a {} hint must be a non-sequence, a "
-          "type, or a TypeConstraint. {} is an instance of "
-          "type.".format(self.string_type, list),
-          e.exception.args[0])
+      if sys.version_info < (3, 9):
+        with self.assertRaises(TypeError) as e:
+          self.beam_type[list]
+        self.assertEqual(
+            "Parameter to a {} hint must be a non-sequence, a "
+            "type, or a TypeConstraint. {} is an instance of "
+            "type.".format(self.string_type, list),
+            e.exception.args[0])
+      else:
+        try:
+          self.beam_type[list]
+        except TypeError:
+          self.fail("built-in composite raised TypeError unexpectedly")
 
     def test_compatibility(self):
       hint1 = self.beam_type[typehints.List[str]]