Replace typing.ByteString with collections.abc.Buffer

sdks/python/apache_beam/coders/coders_property_based_test.py

@@ -48,10 +48,15 @@
 from apache_beam.typehints.schemas import typing_to_runner_api
 from apache_beam.utils.timestamp import Timestamp
 
+try:
+  from collections.abc import Buffer
+except ImportError:
+  from typing_extensions import Buffer
+
 SCHEMA_TYPES_TO_STRATEGY = {
     str: st.text(),
     bytes: st.binary(),
-    typing.ByteString: st.binary(),
+    Buffer: st.binary(),
     # Maximum datetime on year 3000 to conform to Windows OS limits.
     Timestamp: st.datetimes(
         min_value=datetime(1970, 1, 1, 1, 1),

sdks/python/apache_beam/io/external/xlang_snowflakeio_it_test.py

@@ -45,7 +45,6 @@
 import binascii
 import logging
 import unittest
-from typing import ByteString
 from typing import NamedTuple
 
 import apache_beam as beam
@@ -59,6 +58,11 @@
 from apache_beam.testing.util import assert_that
 from apache_beam.testing.util import equal_to
 
+try:
+  from collections.abc import Buffer
+except ImportError:
+  from typing_extensions import Buffer
+
 # pylint: disable=wrong-import-order, wrong-import-position, ungrouped-imports
 try:
   from apache_beam.io.gcp.gcsfilesystem import GCSFileSystem
@@ -82,7 +86,7 @@
     [
         ('number_column', int),
         ('boolean_column', bool),
-        ('bytes_column', ByteString),
+        ('bytes_column', Buffer),
     ])
 
 coders.registry.register_coder(TestRow, coders.RowCoder)

sdks/python/apache_beam/io/gcp/bigquery_schema_tools.py

@@ -22,8 +22,8 @@
 """
 
 import datetime
+from collections.abc import Sequence
 from typing import Optional
-from typing import Sequence
 
 import numpy as np
 

sdks/python/apache_beam/io/gcp/bigquery_schema_tools_test.py

@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import collections
 import logging
 import typing
 import unittest.mock
@@ -50,7 +51,7 @@ def test_check_schema_conversions(self):
         usertype.__annotations__,
         {
             'stn': typing.Optional[str],
-            'temp': typing.Sequence[np.float64],
+            'temp': collections.abc.Sequence[np.float64],
             'count': typing.Optional[np.int64]
         })
 
@@ -158,7 +159,7 @@ def test_check_schema_conversions_with_timestamp(self):
         usertype.__annotations__,
         {
             'stn': typing.Optional[str],
-            'temp': typing.Sequence[np.float64],
+            'temp': collections.abc.Sequence[np.float64],
             'times': typing.Optional[apache_beam.utils.timestamp.Timestamp]
         })
 
@@ -293,7 +294,7 @@ def test_geography_type_support(self):
 
     expected_annotations = {
         'location': typing.Optional[str],
-        'locations': typing.Sequence[str],
+        'locations': collections.abc.Sequence[str],
         'required_location': str
     }
 
@@ -320,7 +321,7 @@ def test_geography_field_type_conversion(self):
 
     # Test repeated GEOGRAPHY field
     result = bq_field_to_type("GEOGRAPHY", "REPEATED")
-    self.assertEqual(result, typing.Sequence[str])
+    self.assertEqual(result, collections.abc.Sequence[str])
 
     # Test GEOGRAPHY field with None mode (should default to nullable)
     result = bq_field_to_type("GEOGRAPHY", None)
@@ -425,7 +426,7 @@ def test_bq_field_to_type_with_overrides(self):
         typing.Optional[datetime.date])
     self.assertEqual(
         bq_field_to_type("DATE", "REPEATED", overrides),
-        typing.Sequence[datetime.date])
+        collections.abc.Sequence[datetime.date])
 
   def test_bq_field_to_type_overrides_can_use_str(self):
     """Test that type_overrides can map DATE/DATETIME/JSON to str."""

sdks/python/apache_beam/typehints/row_type.py

@@ -21,11 +21,9 @@
 from __future__ import annotations
 
 import dataclasses
+from collections.abc import Sequence
 from typing import Any
-from typing import Dict
 from typing import Optional
-from typing import Sequence
-from typing import Tuple
 
 from apache_beam.typehints import typehints
 from apache_beam.typehints.native_type_compatibility import match_dataclass_for_row
@@ -49,11 +47,11 @@ def _user_type_is_generated(user_type: type) -> bool:
 class RowTypeConstraint(typehints.TypeConstraint):
   def __init__(
       self,
-      fields: Sequence[Tuple[str, type]],
+      fields: Sequence[tuple[str, type]],
       user_type,
-      schema_options: Optional[Sequence[Tuple[str, Any]]] = None,
-      field_options: Optional[Dict[str, Sequence[Tuple[str, Any]]]] = None,
-      field_descriptions: Optional[Dict[str, str]] = None):
+      schema_options: Optional[Sequence[tuple[str, Any]]] = None,
+      field_options: Optional[dict[str, Sequence[tuple[str, Any]]]] = None,
+      field_descriptions: Optional[dict[str, str]] = None):
     """For internal use only, no backwards comatibility guaratees.  See
     https://beam.apache.org/documentation/programming-guide/#schemas-for-pl-types
     for guidance on creating PCollections with inferred schemas.
@@ -103,8 +101,8 @@ def __init__(
   @staticmethod
   def from_user_type(
       user_type: type,
-      schema_options: Optional[Sequence[Tuple[str, Any]]] = None,
-      field_options: Optional[Dict[str, Sequence[Tuple[str, Any]]]] = None
+      schema_options: Optional[Sequence[tuple[str, Any]]] = None,
+      field_options: Optional[dict[str, Sequence[tuple[str, Any]]]] = None
   ) -> Optional[RowTypeConstraint]:
     if match_is_named_tuple(user_type):
       fields = [(name, user_type.__annotations__[name])
@@ -136,12 +134,12 @@ def from_user_type(
 
   @staticmethod
   def from_fields(
-      fields: Sequence[Tuple[str, type]],
+      fields: Sequence[tuple[str, type]],
       schema_id: Optional[str] = None,
-      schema_options: Optional[Sequence[Tuple[str, Any]]] = None,
-      field_options: Optional[Dict[str, Sequence[Tuple[str, Any]]]] = None,
+      schema_options: Optional[Sequence[tuple[str, Any]]] = None,
+      field_options: Optional[dict[str, Sequence[tuple[str, Any]]]] = None,
       schema_registry: Optional[SchemaTypeRegistry] = None,
-      field_descriptions: Optional[Dict[str, str]] = None,
+      field_descriptions: Optional[dict[str, str]] = None,
   ) -> RowTypeConstraint:
     return GeneratedClassRowTypeConstraint(
         fields,
@@ -218,10 +216,10 @@ def __init__(
       self,
       fields,
       schema_id: Optional[str] = None,
-      schema_options: Optional[Sequence[Tuple[str, Any]]] = None,
-      field_options: Optional[Dict[str, Sequence[Tuple[str, Any]]]] = None,
+      schema_options: Optional[Sequence[tuple[str, Any]]] = None,
+      field_options: Optional[dict[str, Sequence[tuple[str, Any]]]] = None,
       schema_registry: Optional[SchemaTypeRegistry] = None,
-      field_descriptions: Optional[Dict[str, str]] = None,
+      field_descriptions: Optional[dict[str, str]] = None,
   ):
     from apache_beam.typehints.schemas import named_fields_to_schema
     from apache_beam.typehints.schemas import named_tuple_from_schema

sdks/python/apache_beam/typehints/schemas.py

@@ -32,13 +32,13 @@
   bool          <-----> BOOLEAN
   str           <-----> STRING
   bytes         <-----> BYTES
-  ByteString    ------> BYTES
+  Buffer        ------> BYTES
   Timestamp     <-----> LogicalType(urn="beam:logical_type:micros_instant:v1")
   datetime.date <-----> LogicalType(urn="beam:logical_type:date:v1")
   datetime.time <-----> LogicalType(urn="beam:logical_type:time:v1")
   Decimal       <-----> LogicalType(urn="beam:logical_type:fixed_decimal:v1")
-  Tuple[T, ...] <-----> LogicalType(urn="beam:logical_type:var_tuple:v1")
-  Tuple[T1, T2] <-----> LogicalType(urn="beam:logical_type:fixed_tuple:v1")
+  tuple[T, ...] <-----> LogicalType(urn="beam:logical_type:var_tuple:v1")
+  tuple[T1, T2] <-----> LogicalType(urn="beam:logical_type:fixed_tuple:v1")
   Mapping       <-----> MapType
   Sequence      <-----> ArrayType
   NamedTuple    <-----> RowType
@@ -74,17 +74,13 @@
 import datetime
 import decimal
 import logging
+from collections.abc import Iterable
+from collections.abc import Mapping
+from collections.abc import Sequence
 from typing import Any
-from typing import ByteString
-from typing import Dict
 from typing import Generic
-from typing import Iterable
-from typing import List
-from typing import Mapping
 from typing import NamedTuple
 from typing import Optional
-from typing import Sequence
-from typing import Tuple
 from typing import TypeVar
 from typing import Union
 
@@ -109,6 +105,12 @@
 from apache_beam.utils.python_callable import PythonCallableWithSource
 from apache_beam.utils.timestamp import Timestamp
 
+# Simplify to collections.abc implementation once Python 3.11 is deprecated.
+try:
+  from collections.abc import Buffer
+except ImportError:
+  from typing_extensions import Buffer
+
 PYTHON_ANY_URN = "beam:logical:pythonsdk_any:v1"
 _PYTHON_ANY_FIELD_TYPE_BYTE = "_pythonsdk_any_type_byte"
 _PYTHON_ANY_FIELD_PAYLOAD = "payload"
@@ -134,7 +136,7 @@
 
 # One-way mappings
 PRIMITIVE_TO_ATOMIC_TYPE.update({
-    ByteString: schema_pb2.BYTES,
+    Buffer: schema_pb2.BYTES,
     # Allow users to specify a native int, and use INT64 as the cross-language
     # representation. Technically ints have unlimited precision, but RowCoder
     # should throw an error if it sees one with a bit width > 64 when encoding.
@@ -149,12 +151,12 @@
 
 
 def named_fields_to_schema(
-    names_and_types: Union[Dict[str, type], Sequence[Tuple[str, type]]],
+    names_and_types: Union[dict[str, type], Sequence[tuple[str, type]]],
     schema_id: Optional[str] = None,
-    schema_options: Optional[Sequence[Tuple[str, Any]]] = None,
-    field_options: Optional[Dict[str, Sequence[Tuple[str, Any]]]] = None,
+    schema_options: Optional[Sequence[tuple[str, Any]]] = None,
+    field_options: Optional[dict[str, Sequence[tuple[str, Any]]]] = None,
     schema_registry: SchemaTypeRegistry = SCHEMA_REGISTRY,
-    field_descriptions: Optional[Dict[str, str]] = None,
+    field_descriptions: Optional[dict[str, str]] = None,
 ):
   schema_options = schema_options or []
   field_options = field_options or {}
@@ -200,7 +202,7 @@ def named_fields_to_schema(
 
 
 def named_fields_from_schema(
-    schema):  # (schema_pb2.Schema) -> typing.List[typing.Tuple[str, type]]
+    schema):  # (schema_pb2.Schema) -> list[typing.tuple[str, type]]
   return [(field.name, typing_from_runner_api(field.type))
           for field in schema.fields]
 
@@ -240,15 +242,15 @@ def value_from_runner_api(
 
 
 def option_to_runner_api(
-    option: Tuple[str, Any],
+    option: tuple[str, Any],
     schema_registry: SchemaTypeRegistry = SCHEMA_REGISTRY) -> schema_pb2.Option:
   return SchemaTranslation(
       schema_registry=schema_registry).option_to_runner_api(option)
 
 
 def option_from_runner_api(
     option_proto: schema_pb2.Option,
-    schema_registry: SchemaTypeRegistry = SCHEMA_REGISTRY) -> Tuple[str, Any]:
+    schema_registry: SchemaTypeRegistry = SCHEMA_REGISTRY) -> tuple[str, Any]:
   return SchemaTranslation(
       schema_registry=schema_registry).option_from_runner_api(option_proto)
 
@@ -540,14 +542,14 @@ def value_to_runner_api(self, typing_proto: schema_pb2.FieldType, value):
           f"{typing_proto!r}.")
 
   def option_from_runner_api(
-      self, option_proto: schema_pb2.Option) -> Tuple[str, Any]:
+      self, option_proto: schema_pb2.Option) -> tuple[str, Any]:
     if not option_proto.HasField('type'):
       return option_proto.name, None
 
     value = self.value_from_runner_api(option_proto.type, option_proto.value)
     return option_proto.name, value
 
-  def option_to_runner_api(self, option: Tuple[str, Any]) -> schema_pb2.Option:
+  def option_to_runner_api(self, option: tuple[str, Any]) -> schema_pb2.Option:
     name, value = option
 
     if value is None:
@@ -732,7 +734,7 @@ def schema_from_element_type(element_type: type) -> schema_pb2.Schema:
 
 
 def named_fields_from_element_type(
-    element_type: type) -> List[Tuple[str, type]]:
+    element_type: type) -> list[tuple[str, type]]:
   return named_fields_from_schema(schema_from_element_type(element_type))
 
 
@@ -1539,7 +1541,7 @@ def _from_typing(cls, typ):
     return cls()
 
 
-_TUPLE_NAMEDTUPLE_CACHE: Dict[int, type] = {}
+_TUPLE_NAMEDTUPLE_CACHE: dict[int, type] = {}
 
 
 def _get_tuple_namedtuple(n: int) -> type:
@@ -1564,7 +1566,7 @@ def urn(cls):
   def language_type(self=None):
     if self is None or not self._tuple_types:
       return tuple
-    return Tuple[self._tuple_types]
+    return tuple[self._tuple_types]
 
   def representation_type(self):
     if not self._tuple_types:
@@ -1609,7 +1611,7 @@ def urn(cls):
   def language_type(self=None):
     if self is None or self._elem_type is object:
       return tuple
-    return Tuple[self._elem_type, ...]
+    return tuple[self._elem_type, ...]
 
   def representation_type(self):
     return Sequence[self._elem_type]