[Gemini] Migrate all remaining uses of typing types with built-in equivalents

.github/trigger_files/beam_PostCommit_Python.json

@@ -1,5 +1,5 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run.",
   "pr": "38069",
-  "modification": 40
+  "modification": 41
 }

sdks/python/apache_beam/coders/coder_impl.py

@@ -30,6 +30,7 @@
 """
 # pytype: skip-file
 
+# ruff: noqa: UP006
 import dataclasses
 import decimal
 import enum

sdks/python/apache_beam/coders/coders.py

@@ -43,13 +43,9 @@
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Callable
-from typing import Dict
 from typing import Iterable
-from typing import List
 from typing import Optional
 from typing import Sequence
-from typing import Tuple
-from typing import Type
 from typing import TypeVar
 from typing import overload
 
@@ -68,7 +64,6 @@
 from apache_beam.utils import windowed_value
 
 if TYPE_CHECKING:
-  from apache_beam.coders.typecoders import CoderRegistry
   from apache_beam.runners.pipeline_context import PipelineContext
 
 # pylint: disable=wrong-import-order, wrong-import-position, ungrouped-imports
@@ -122,7 +117,7 @@
 T = TypeVar('T')
 CoderT = TypeVar('CoderT', bound='Coder')
 ProtoCoderT = TypeVar('ProtoCoderT', bound='ProtoCoder')
-ConstructorFn = Callable[[Optional[Any], List['Coder'], 'PipelineContext'], Any]
+ConstructorFn = Callable[[Optional[Any], list['Coder'], 'PipelineContext'], Any]
 
 
 def serialize_coder(coder):
@@ -1508,7 +1503,7 @@ def __hash__(self):
 
 class _OrderedUnionCoder(FastCoder):
   def __init__(
-      self, *coder_types: Tuple[type, Coder], fallback_coder: Optional[Coder]):
+      self, *coder_types: tuple[type, Coder], fallback_coder: Optional[Coder]):
     self._coder_types = coder_types
     self._fallback_coder = fallback_coder
 
@@ -1816,7 +1811,7 @@ def _create_impl(self):
   def to_type_hint(self):
     return self._window_coder.to_type_hint()
 
-  def _get_component_coders(self) -> List[Coder]:
+  def _get_component_coders(self) -> list[Coder]:
     return [self._window_coder]
 
   def is_deterministic(self) -> bool:

sdks/python/apache_beam/coders/coders_test_common.py

@@ -31,7 +31,6 @@
 import unittest
 from decimal import Decimal
 from typing import Any
-from typing import List
 from typing import NamedTuple
 
 import pytest
@@ -145,7 +144,7 @@ class CodersTest(unittest.TestCase):
   # nested and unnested context.
 
   # Common test values representing Python's built-in types.
-  test_values_deterministic: List[Any] = [
+  test_values_deterministic: list[Any] = [
       None,
       1,
       -1,

sdks/python/apache_beam/coders/observable_test.py

@@ -20,7 +20,6 @@
 
 import logging
 import unittest
-from typing import List
 from typing import Optional
 
 from apache_beam.coders import observable
@@ -29,7 +28,7 @@
 class ObservableMixinTest(unittest.TestCase):
   observed_count = 0
   observed_sum = 0
-  observed_keys: List[Optional[str]] = []
+  observed_keys: list[Optional[str]] = []
 
   def observer(self, value, key=None):
     self.observed_count += 1

sdks/python/apache_beam/coders/row_coder_test.py

@@ -43,7 +43,7 @@
         ("name", str),
         ("age", np.int32),
         ("address", typing.Optional[str]),
-        ("aliases", typing.List[str]),
+        ("aliases", list[str]),
         ("knows_javascript", bool),
         ("payload", typing.Optional[bytes]),
         ("custom_metadata", typing.Mapping[str, int]),
@@ -53,7 +53,7 @@
 NullablePerson = typing.NamedTuple(
     "NullablePerson",
     [("name", typing.Optional[str]), ("age", np.int32),
-     ("address", typing.Optional[str]), ("aliases", typing.List[str]),
+     ("address", typing.Optional[str]), ("aliases", list[str]),
      ("knows_javascript", bool), ("payload", typing.Optional[bytes]),
      ("custom_metadata", typing.Mapping[str, int]),
      ("favorite_time", typing.Optional[Timestamp]),

sdks/python/apache_beam/coders/slow_stream.py

@@ -22,15 +22,14 @@
 # pytype: skip-file
 
 import struct
-from typing import List
 
 
 class OutputStream(object):
   """For internal use only; no backwards-compatibility guarantees.
 
   A pure Python implementation of stream.OutputStream."""
   def __init__(self):
-    self.data: List[bytes] = []
+    self.data: list[bytes] = []
     self.byte_count = 0
 
   def write(self, b: bytes, nested: bool = False) -> None:

sdks/python/apache_beam/coders/standard_coders_test.py

@@ -26,8 +26,6 @@
 import sys
 import unittest
 from copy import deepcopy
-from typing import Dict
-from typing import Tuple
 
 import numpy as np
 import yaml
@@ -283,7 +281,7 @@ def json_value_parser(self, coder_spec):
   # Used when --fix is passed.
 
   fix = False
-  to_fix: Dict[Tuple[int, bytes], bytes] = {}
+  to_fix: dict[tuple[int, bytes], bytes] = {}
 
   @classmethod
   def tearDownClass(cls):

sdks/python/apache_beam/coders/typecoders.py

@@ -66,10 +66,7 @@ def MakeXyzs(v):
 
 # pytype: skip-file
 from typing import Any
-from typing import Dict
 from typing import Iterable
-from typing import List
-from typing import Type
 
 from apache_beam.coders import coders
 from apache_beam.typehints import typehints
@@ -81,8 +78,8 @@ def MakeXyzs(v):
 class CoderRegistry(object):
   """A coder registry for typehint/coder associations."""
   def __init__(self, fallback_coder=None):
-    self._coders: Dict[Any, Type[coders.Coder]] = {}
-    self.custom_types: List[Any] = []
+    self._coders: dict[Any, type[coders.Coder]] = {}
+    self.custom_types: list[Any] = []
     self.register_standard_coders(fallback_coder)
 
   def register_standard_coders(self, fallback_coder):
@@ -110,7 +107,7 @@ def register_fallback_coder(self, fallback_coder):
 
   def _register_coder_internal(
       self, typehint_type: Any,
-      typehint_coder_class: Type[coders.Coder]) -> None:
+      typehint_coder_class: type[coders.Coder]) -> None:
     self._coders[typehint_type] = typehint_coder_class
 
   @staticmethod
@@ -123,7 +120,7 @@ def _normalize_typehint_type(typehint_type):
 
   def register_coder(
       self, typehint_type: Any,
-      typehint_coder_class: Type[coders.Coder]) -> None:
+      typehint_coder_class: type[coders.Coder]) -> None:
     """
     Register a user type with a coder.
 
@@ -244,7 +241,7 @@ class FirstOf(object):
   """For internal use only; no backwards-compatibility guarantees.
 
   A class used to get the first matching coder from a list of coders."""
-  def __init__(self, coders: Iterable[Type[coders.Coder]]) -> None:
+  def __init__(self, coders: Iterable[type[coders.Coder]]) -> None:
     self._coders = coders
 
   def from_type_hint(self, typehint, registry):

sdks/python/apache_beam/dataframe/frame_base.py

@@ -25,7 +25,6 @@
 from inspect import unwrap
 from typing import Any
 from typing import Optional
-from typing import Tuple
 from typing import Union
 
 import pandas as pd
@@ -163,7 +162,7 @@ def binop(self, other):
 DeferredBase._pandas_type_map[None] = _DeferredScalar
 
 
-def name_and_func(method: Union[str, Callable]) -> Tuple[str, Callable]:
+def name_and_func(method: Union[str, Callable]) -> tuple[str, Callable]:
   """For the given method name or method, return the method name and the method
   itself.
 

sdks/python/apache_beam/dataframe/schemas_test.py

@@ -64,57 +64,36 @@ def check_df_pcoll_equal(actual):
 #   pd.Series([b'abc'], dtype=bytes).dtype != 'S'
 #   pd.Series([b'abc'], dtype=bytes).astype(bytes).dtype == 'S'
 # (test data, pandas_type, column_name, beam_type)
-COLUMNS: typing.List[typing.Tuple[typing.List[typing.Any],
-                                  typing.Any,
-                                  str,
-                                  typing.Any]] = [
-                                      ([375, 24, 0, 10, 16],
-                                       np.int32,
-                                       'i32',
-                                       np.int32),
-                                      ([375, 24, 0, 10, 16],
-                                       np.int64,
-                                       'i64',
-                                       np.int64),
-                                      ([375, 24, None, 10, 16],
-                                       pd.Int32Dtype(),
-                                       'i32_nullable',
-                                       typing.Optional[np.int32]),
-                                      ([375, 24, None, 10, 16],
-                                       pd.Int64Dtype(),
-                                       'i64_nullable',
-                                       typing.Optional[np.int64]),
-                                      ([375., 24., None, 10., 16.],
-                                       np.float64,
-                                       'f64',
-                                       typing.Optional[np.float64]),
-                                      ([375., 24., None, 10., 16.],
-                                       np.float32,
-                                       'f32',
-                                       typing.Optional[np.float32]),
-                                      ([True, False, True, True, False],
-                                       bool,
-                                       'bool',
-                                       bool),
-                                      (['Falcon', 'Ostrich', None, 3.14, 0],
-                                       object,
-                                       'any',
-                                       typing.Any),
-                                      ([True, False, True, None, False],
-                                       pd.BooleanDtype(),
-                                       'bool_nullable',
-                                       typing.Optional[bool]),
-                                      ([
-                                          'Falcon',
-                                          'Ostrich',
-                                          None,
-                                          'Aardvark',
-                                          'Elephant'
-                                      ],
-                                       pd.StringDtype(),
-                                       'strdtype',
-                                       typing.Optional[str]),
-                                  ]
+COLUMNS: list[tuple[list[typing.Any], typing.Any, str, typing.Any]] = [
+    ([375, 24, 0, 10, 16], np.int32, 'i32', np.int32),
+    ([375, 24, 0, 10, 16], np.int64, 'i64', np.int64),
+    ([375, 24, None, 10, 16],
+     pd.Int32Dtype(),
+     'i32_nullable',
+     typing.Optional[np.int32]),
+    ([375, 24, None, 10, 16],
+     pd.Int64Dtype(),
+     'i64_nullable',
+     typing.Optional[np.int64]),
+    ([375., 24., None, 10., 16.],
+     np.float64,
+     'f64',
+     typing.Optional[np.float64]),
+    ([375., 24., None, 10., 16.],
+     np.float32,
+     'f32',
+     typing.Optional[np.float32]),
+    ([True, False, True, True, False], bool, 'bool', bool),
+    (['Falcon', 'Ostrich', None, 3.14, 0], object, 'any', typing.Any),
+    ([True, False, True, None, False],
+     pd.BooleanDtype(),
+     'bool_nullable',
+     typing.Optional[bool]),
+    (['Falcon', 'Ostrich', None, 'Aardvark', 'Elephant'],
+     pd.StringDtype(),
+     'strdtype',
+     typing.Optional[str]),
+]
 
 NICE_TYPES_DF = pd.DataFrame(columns=[name for _, _, name, _ in COLUMNS])
 for arr, dtype, name, _ in COLUMNS:
@@ -125,9 +104,7 @@ def check_df_pcoll_equal(actual):
 SERIES_TESTS = [(pd.Series(arr, dtype=dtype, name=name), arr, beam_type)
                 for (arr, dtype, name, beam_type) in COLUMNS]
 
-_TEST_ARRAYS: typing.List[typing.List[typing.Any]] = [
-    arr for (arr, _, _, _) in COLUMNS
-]
+_TEST_ARRAYS: list[list[typing.Any]] = [arr for (arr, _, _, _) in COLUMNS]
 DF_RESULT = list(zip(*_TEST_ARRAYS))
 BEAM_SCHEMA = typing.NamedTuple(  # type: ignore
     'BEAM_SCHEMA', [(name, beam_type) for _, _, name, beam_type in COLUMNS])

sdks/python/apache_beam/examples/inference/multi_language_inference/multi_language_custom_transform/multi_language_custom_transform/composite_transform.py

@@ -15,7 +15,6 @@
 #
 
 import logging
-import signal
 import typing
 
 import apache_beam as beam
@@ -84,7 +83,7 @@ def __init__(self, bert_tokenizer):
       self.bert_tokenizer = bert_tokenizer
       logging.info('Starting Postprocess')
 
-    def process(self, element: typing.Tuple[str, PredictionResult]) \
+    def process(self, element: tuple[str, PredictionResult]) \
         -> typing.Iterable[str]:
       text, prediction_result = element
       inputs = prediction_result.example

sdks/python/apache_beam/examples/inference/tfx_bsl/build_tensorflow_model.py

@@ -17,7 +17,6 @@
 
 # Intended only for internal testing.
 
-from typing import Dict
 from typing import Optional
 
 import tensorflow as tf
@@ -114,7 +113,7 @@ def save_tf_model_with_signature(
     model=None,
     preprocess_input=None,
     input_dtype=tf.float32,
-    feature_description: Optional[Dict] = None,
+    feature_description: Optional[dict] = None,
     **kwargs,
 ):
   """