Clean up unnecessary Python version check, remove old typing types from internal directory

sdks/python/apache_beam/internal/dill_pickler.py

@@ -39,8 +39,6 @@
 import types
 import zlib
 from typing import Any
-from typing import Dict
-from typing import Tuple
 
 import dill
 
@@ -50,7 +48,7 @@
 
 settings = {'dill_byref': None}
 
-patch_save_code = sys.version_info >= (3, 10) and dill.__version__ == "0.3.1.1"
+patch_save_code = dill.__version__ == "0.3.1.1"
 
 if patch_save_code:
   # The following function is based on 'save_code' from 'dill'
@@ -315,7 +313,7 @@ def save_module(pickler, obj):
   # Pickle module dictionaries (commonly found in lambda's globals)
   # by referencing their module.
   old_save_module_dict = dill.dill.save_module_dict
-  known_module_dicts: Dict[int, Tuple[types.ModuleType, Dict[str, Any]]] = {}
+  known_module_dicts: dict[int, tuple[types.ModuleType, dict[str, Any]]] = {}
 
   @dill.dill.register(dict)
   def new_save_module_dict(pickler, obj):

sdks/python/apache_beam/internal/metrics/metric.py

@@ -30,9 +30,7 @@
 import threading
 import time
 from typing import TYPE_CHECKING
-from typing import Dict
 from typing import Optional
-from typing import Type
 from typing import Union
 
 from apache_beam.metrics import monitoring_infos
@@ -59,7 +57,7 @@ class Metrics(object):
   @staticmethod
   def counter(
       urn: str,
-      labels: Optional[Dict[str, str]] = None,
+      labels: Optional[dict[str, str]] = None,
       process_wide: bool = False) -> UserMetrics.DelegatingCounter:
     """Obtains or creates a Counter metric.
 
@@ -82,22 +80,22 @@ def counter(
 class MetricLogger(object):
   """Simple object to locally aggregate and log metrics."""
   def __init__(self) -> None:
-    self._metric: Dict[MetricName, 'MetricCell'] = {}
+    self._metric: dict[MetricName, 'MetricCell'] = {}
     self._lock = threading.Lock()
     self._last_logging_millis = int(time.time() * 1000)
     self.minimum_logging_frequency_msec = 180000
 
   def update(
       self,
-      cell_type: Union[Type['MetricCell'], 'MetricCellFactory'],
+      cell_type: Union[type['MetricCell'], 'MetricCellFactory'],
       metric_name: MetricName,
       value: object) -> None:
     cell = self._get_metric_cell(cell_type, metric_name)
     cell.update(value)
 
   def _get_metric_cell(
       self,
-      cell_type: Union[Type['MetricCell'], 'MetricCellFactory'],
+      cell_type: Union[type['MetricCell'], 'MetricCellFactory'],
       metric_name: MetricName) -> 'MetricCell':
     with self._lock:
       if metric_name not in self._metric:
@@ -139,7 +137,7 @@ class ServiceCallMetric(object):
   def __init__(
       self,
       request_count_urn: str,
-      base_labels: Optional[Dict[str, str]] = None) -> None:
+      base_labels: Optional[dict[str, str]] = None) -> None:
     self.base_labels = base_labels if base_labels else {}
     self.request_count_urn = request_count_urn
 

sdks/python/apache_beam/internal/util.py

@@ -25,13 +25,9 @@
 import logging
 import threading
 import weakref
+from collections.abc import Iterable
 from multiprocessing.pool import ThreadPool
 from typing import Any
-from typing import Dict
-from typing import Iterable
-from typing import List
-from typing import Tuple
-from typing import Type
 from typing import TypeVar
 from typing import Union
 
@@ -68,9 +64,9 @@ def __hash__(self):
 
 def remove_objects_from_args(
     args: Iterable[Any],
-    kwargs: Dict[str, Any],
-    pvalue_class: Union[Type[T], Tuple[Type[T], ...]]
-) -> Tuple[List[Any], Dict[str, Any], List[T]]:
+    kwargs: dict[str, Any],
+    pvalue_class: Union[type[T], tuple[type[T], ...]]
+) -> tuple[list[Any], dict[str, Any], list[T]]:
   """For internal use only; no backwards-compatibility guarantees.
 
   Replaces all objects of a given type in args/kwargs with a placeholder.