Remove 3.9-specific branches in typehinting module

sdks/python/apache_beam/typehints/native_type_compatibility.py

@@ -22,7 +22,6 @@
 import collections
 import collections.abc
 import logging
-import sys
 import types
 import typing
 from typing import Generic
@@ -328,8 +327,7 @@ def convert_to_beam_type(typ):
   # pipe operator as Union and types.UnionType are introduced
   # in Python 3.10.
   # GH issue: https://github.com/apache/beam/issues/21972
-  if (sys.version_info.major == 3 and
-      sys.version_info.minor >= 10) and (isinstance(typ, types.UnionType)):
+  if isinstance(typ, types.UnionType):
     typ = typing.Union[typ]
 
   if getattr(typ, '__module__', None) == 'typing':
@@ -352,7 +350,7 @@ def convert_to_beam_type(typ):
     # TODO(https://github.com/apache/beam/issues/19954): Currently unhandled.
     _LOGGER.info('Converting string literal type hint to Any: "%s"', typ)
     return typehints.Any
-  elif sys.version_info >= (3, 10) and isinstance(typ, typing.NewType):  # pylint: disable=isinstance-second-argument-not-valid-type
+  elif isinstance(typ, typing.NewType):  # pylint: disable=isinstance-second-argument-not-valid-type
     # Special case for NewType, where, since Python 3.10, NewType is now a class
     # rather than a function.
     # TODO(https://github.com/apache/beam/issues/20076): Currently unhandled.

sdks/python/apache_beam/typehints/trivial_inference.py

@@ -394,13 +394,7 @@ def infer_return_type_func(f, input_types, debug=False, depth=0):
   inst_size = 2
   opt_arg_size = 0
 
-  # Python 3.10: bpo-27129 changes jump offsets to use instruction offsets,
-  # not byte offsets. The offsets were halved (16 bits fro instructions vs 8
-  # bits for bytes), so we have to double the value of arg.
-  if (sys.version_info.major, sys.version_info.minor) >= (3, 10):
-    jump_multiplier = 2
-  else:
-    jump_multiplier = 1
+  jump_multiplier = 2
 
   last_pc = -1
   last_real_opname = opname = None

sdks/python/apache_beam/typehints/typehints.py

@@ -67,7 +67,6 @@
 
 import copy
 import logging
-import sys
 import types
 import typing
 from collections import abc
@@ -392,9 +391,8 @@ def validate_composite_type_param(type_param, error_msg_prefix):
       not isinstance(type_param, tuple(possible_classes)) and
       type_param is not None and
       getattr(type_param, '__module__', None) != 'typing')
-  if sys.version_info.major == 3 and sys.version_info.minor >= 10:
-    if isinstance(type_param, types.UnionType):
-      is_not_type_constraint = False
+  if isinstance(type_param, types.UnionType):
+    is_not_type_constraint = False
 
   if is_not_type_constraint:
     raise TypeError(

sdks/python/apache_beam/typehints/typehints_test.py

@@ -22,7 +22,6 @@
 import collections.abc
 import functools
 import re
-import sys
 import typing
 import unittest
 
@@ -1929,12 +1928,11 @@ def expand(self, pcoll: typing.Any) -> typehints.Any:
   def test_pipe_operator_as_union(self):
     # union types can be written using pipe operator from Python 3.10.
     # https://peps.python.org/pep-0604/
-    if sys.version_info.major == 3 and sys.version_info.minor >= 10:
-      type_a = int | float  # pylint: disable=unsupported-binary-operation
-      type_b = typing.Union[int, float]
-      self.assertEqual(
-          native_type_compatibility.convert_to_beam_type(type_a),
-          native_type_compatibility.convert_to_beam_type(type_b))
+    type_a = int | float  # pylint: disable=unsupported-binary-operation
+    type_b = typing.Union[int, float]
+    self.assertEqual(
+        native_type_compatibility.convert_to_beam_type(type_a),
+        native_type_compatibility.convert_to_beam_type(type_b))
 
 
 class TestNonBuiltInGenerics(unittest.TestCase):