[BEAM-12465] Fix nested subscripted Generics

This fix applies to Py3.7+. Version 3.6 was not affected.

sdks/python/apache_beam/typehints/native_type_compatibility_test.py

@@ -40,6 +40,13 @@ class _TestClass(object):
   pass
 
 
+T = typing.TypeVar('T')
+
+
+class _TestGeneric(typing.Generic[T]):
+  pass
+
+
 class NativeTypeCompatibilityTest(unittest.TestCase):
   def test_convert_to_beam_type(self):
     test_cases = [
@@ -64,6 +71,8 @@ def test_convert_to_beam_type(self):
         ('test class', _TestClass, _TestClass),
         ('test class in list', typing.List[_TestClass],
          typehints.List[_TestClass]),
+        ('generic bare', _TestGeneric, _TestGeneric),
+        ('generic subscripted', _TestGeneric[int], _TestGeneric[int]),
         ('complex tuple', typing.Tuple[bytes, typing.List[typing.Tuple[
             bytes, typing.Union[int, bytes, float]]]],
          typehints.Tuple[bytes, typehints.List[typehints.Tuple[
@@ -84,6 +93,10 @@ def test_convert_to_beam_type(self):
                          typehints.TypeVariable('V')]),
         ('iterator', typing.Iterator[typing.Any],
          typehints.Iterator[typehints.Any]),
+        ('nested generic bare', typing.List[_TestGeneric],
+         typehints.List[_TestGeneric]),
+        ('nested generic subscripted', typing.List[_TestGeneric[int]],
+         typehints.List[_TestGeneric[int]]),
     ]
 
     for test_case in test_cases:

sdks/python/apache_beam/typehints/typehints.py

@@ -318,6 +318,19 @@ def __getitem___(self, py_type):
     raise NotImplementedError
 
 
+def is_typing_generic(type_param):
+  """Determines if an object is a subscripted typing.Generic type, such as
+  PCollection[int].
+
+  Such objects are considered valid type parameters.
+
+  Always returns false for Python versions below 3.7.
+  """
+  if hasattr(typing, '_GenericAlias'):
+    return isinstance(type_param, typing._GenericAlias)
+  return False
+
+
 def validate_composite_type_param(type_param, error_msg_prefix):
   """Determines if an object is a valid type parameter to a
   :class:`CompositeTypeHint`.
@@ -338,6 +351,7 @@ def validate_composite_type_param(type_param, error_msg_prefix):
   # Must either be a TypeConstraint instance or a basic Python type.
   possible_classes = [type, TypeConstraint]
   is_not_type_constraint = (
+      not is_typing_generic(type_param) and
       not isinstance(type_param, tuple(possible_classes)) and
       type_param is not None and
       getattr(type_param, '__module__', None) != 'typing')