Handle types.GenericAlias types properly during validation (#27633)

Author: jrmccluskey

sdks/python/apache_beam/typehints/native_type_compatibility_test.py

@@ -49,6 +49,11 @@ class _TestGeneric(typing.Generic[T]):
   pass
 
 
+class _TestPair(typing.NamedTuple('TestTuple', [('first', T), ('second', T)]),
+                typing.Generic[T]):
+  pass
+
+
 class NativeTypeCompatibilityTest(unittest.TestCase):
   def test_convert_to_beam_type(self):
     test_cases = [
@@ -99,6 +104,8 @@ def test_convert_to_beam_type(self):
          typehints.List[_TestGeneric]),
         ('nested generic subscripted', typing.List[_TestGeneric[int]],
          typehints.List[_TestGeneric[int]]),
+        ('nested generic with any', typing.List[_TestPair[typing.Any]],
+         typehints.List[_TestPair[typing.Any]]),
     ]
 
     for test_case in test_cases:

sdks/python/apache_beam/typehints/typehints.py

@@ -355,11 +355,12 @@ def is_typing_generic(type_param):
 
   Such objects are considered valid type parameters.
 
-  Always returns false for Python versions below 3.7.
+  For Python versions 3.9 and above, also permits types.GenericAlias.
   """
-  if hasattr(typing, '_GenericAlias'):
-    return isinstance(type_param, typing._GenericAlias)
-  return False
+  if hasattr(types, "GenericAlias") and isinstance(type_param,
+                                                   types.GenericAlias):
+    return True
+  return isinstance(type_param, typing._GenericAlias)
 
 
 def validate_composite_type_param(type_param, error_msg_prefix):

sdks/python/apache_beam/typehints/typehints_test.py

@@ -619,6 +619,13 @@ def test_builtin_and_type_compatibility(self):
       self.assertCompatible(list, typing.List)
       self.assertCompatible(list[int], typing.List[int])
 
+  def test_is_typing_generic(self):
+    self.assertTrue(typehints.is_typing_generic(typing.List[str]))
+
+  def test_builtin_is_typing_generic(self):
+    if sys.version_info >= (3, 9):
+      self.assertTrue(typehints.is_typing_generic(list[str]))
+
 
 class KVHintTestCase(TypeHintTestCase):
   def test_getitem_param_must_be_tuple(self):
@@ -791,6 +798,10 @@ def test_getitem_invalid_composite_type_param(self):
         except TypeError:
           self.fail("built-in composite raised TypeError unexpectedly")
 
+    def test_non_typing_generic(self):
+      testCase = DummyTestClass1()
+      self.assertFalse(typehints.is_typing_generic(testCase))
+
     def test_compatibility(self):
       hint1 = self.beam_type[typehints.List[str]]
       hint2 = self.beam_type[typehints.Tuple[int, int]]