diff --git a/sdks/python/apache_beam/typehints/native_type_compatibility_test.py b/sdks/python/apache_beam/typehints/native_type_compatibility_test.py index 8dcff9fc2d7d..e7b69932a08e 100644 --- a/sdks/python/apache_beam/typehints/native_type_compatibility_test.py +++ b/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] diff --git a/sdks/python/apache_beam/typehints/typehints.py b/sdks/python/apache_beam/typehints/typehints.py index 71d56ae4b4f9..7ea05f27cc87 100644 --- a/sdks/python/apache_beam/typehints/typehints.py +++ b/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.' % diff --git a/sdks/python/apache_beam/typehints/typehints_test.py b/sdks/python/apache_beam/typehints/typehints_test.py index d8893a3f474a..1f6b98eccbea 100644 --- a/sdks/python/apache_beam/typehints/typehints_test.py +++ b/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]]