This is an automated email from the ASF dual-hosted git repository.
jrmccluskey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 8c100dd7fe4 Replace typing.ByteString with collections.abc.Buffer
(#40554)
8c100dd7fe4 is described below
commit 8c100dd7fe44ede467baad886181b4df117cf219
Author: Jack McCluskey <[email protected]>
AuthorDate: Thu Oct 15 10:07:28 2026 -0400
Replace typing.ByteString with collections.abc.Buffer (#40554)
* Replace typing.ByteString with collections.abc.Buffer
* yapf
* Replace ByteString in coders_property_based_test.py
* update snowflakeio test to use buffer
* Swap BQ schema tests to collections.abc.Sequence
* corresponding change
* missing import
---
.../coders/coders_property_based_test.py | 7 ++-
.../io/external/xlang_snowflakeio_it_test.py | 8 +++-
.../apache_beam/io/gcp/bigquery_schema_tools.py | 2 +-
.../io/gcp/bigquery_schema_tools_test.py | 11 ++---
sdks/python/apache_beam/typehints/row_type.py | 30 ++++++-------
sdks/python/apache_beam/typehints/schemas.py | 50 +++++++++++-----------
sdks/python/apache_beam/typehints/schemas_test.py | 29 +++++++------
.../apache_beam/typehints/testing/strategies.py | 4 +-
8 files changed, 77 insertions(+), 64 deletions(-)
diff --git a/sdks/python/apache_beam/coders/coders_property_based_test.py
b/sdks/python/apache_beam/coders/coders_property_based_test.py
index 9b5600d7156..aa14a178745 100644
--- a/sdks/python/apache_beam/coders/coders_property_based_test.py
+++ b/sdks/python/apache_beam/coders/coders_property_based_test.py
@@ -48,10 +48,15 @@ from apache_beam.typehints.schemas import
PRIMITIVE_TO_ATOMIC_TYPE
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),
diff --git a/sdks/python/apache_beam/io/external/xlang_snowflakeio_it_test.py
b/sdks/python/apache_beam/io/external/xlang_snowflakeio_it_test.py
index 70293f863f4..f85b77e7a29 100644
--- a/sdks/python/apache_beam/io/external/xlang_snowflakeio_it_test.py
+++ b/sdks/python/apache_beam/io/external/xlang_snowflakeio_it_test.py
@@ -45,7 +45,6 @@ import argparse
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.test_pipeline import TestPipeline
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 @@ TestRow = NamedTuple(
[
('number_column', int),
('boolean_column', bool),
- ('bytes_column', ByteString),
+ ('bytes_column', Buffer),
])
coders.registry.register_coder(TestRow, coders.RowCoder)
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_schema_tools.py
b/sdks/python/apache_beam/io/gcp/bigquery_schema_tools.py
index d3d608b1fc6..bd523835daa 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_schema_tools.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery_schema_tools.py
@@ -22,8 +22,8 @@ NOTHING IN THIS FILE HAS BACKWARDS COMPATIBILITY GUARANTEES.
"""
import datetime
+from collections.abc import Sequence
from typing import Optional
-from typing import Sequence
import numpy as np
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_schema_tools_test.py
b/sdks/python/apache_beam/io/gcp/bigquery_schema_tools_test.py
index 73cedb3a6ae..788f872ced6 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_schema_tools_test.py
+++ b/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 @@ class TestBigQueryToSchema(unittest.TestCase):
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 @@ class TestBigQueryToSchema(unittest.TestCase):
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 @@ class TestBigQueryToSchema(unittest.TestCase):
expected_annotations = {
'location': typing.Optional[str],
- 'locations': typing.Sequence[str],
+ 'locations': collections.abc.Sequence[str],
'required_location': str
}
@@ -320,7 +321,7 @@ class TestBigQueryToSchema(unittest.TestCase):
# 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 @@ class TestTypeOverridesSchemaTools(unittest.TestCase):
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."""
diff --git a/sdks/python/apache_beam/typehints/row_type.py
b/sdks/python/apache_beam/typehints/row_type.py
index 5e0faf91f55..89e15b5df82 100644
--- a/sdks/python/apache_beam/typehints/row_type.py
+++ b/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 @@ class RowTypeConstraint(typehints.TypeConstraint):
@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 @@ class RowTypeConstraint(typehints.TypeConstraint):
@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 @@ class GeneratedClassRowTypeConstraint(RowTypeConstraint):
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
diff --git a/sdks/python/apache_beam/typehints/schemas.py
b/sdks/python/apache_beam/typehints/schemas.py
index 9caccc762f7..a023d4f4cd6 100644
--- a/sdks/python/apache_beam/typehints/schemas.py
+++ b/sdks/python/apache_beam/typehints/schemas.py
@@ -32,13 +32,13 @@ Imposes a mapping between common Python types and Beam
portable schemas
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
@@ -75,17 +75,13 @@ import dataclasses
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
@@ -110,6 +106,12 @@ from apache_beam.utils import proto_utils
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"
@@ -135,7 +137,7 @@ ATOMIC_TYPE_TO_PRIMITIVE = dict((atomic, typ) for typ,
atomic in _PRIMITIVES)
# 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.
@@ -150,12 +152,12 @@ _SCHEMA_ID_CACHE = {}
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 {}
@@ -201,7 +203,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]
@@ -241,7 +243,7 @@ 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)
@@ -249,7 +251,7 @@ def option_to_runner_api(
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)
@@ -541,14 +543,14 @@ class SchemaTranslation(object):
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:
@@ -751,7 +753,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))
@@ -1558,7 +1560,7 @@ class JdbcTimeType(LogicalType[datetime.time,
MillisInstant, str]):
return cls()
-_TUPLE_NAMEDTUPLE_CACHE: Dict[int, type] = {}
+_TUPLE_NAMEDTUPLE_CACHE: dict[int, type] = {}
def _get_tuple_namedtuple(n: int) -> type:
@@ -1583,7 +1585,7 @@ class FixedTupleLogicalType(NoArgumentLogicalType[tuple,
Any]):
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:
@@ -1628,7 +1630,7 @@ class VarTupleLogicalType(NoArgumentLogicalType[tuple,
Sequence]):
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]
diff --git a/sdks/python/apache_beam/typehints/schemas_test.py
b/sdks/python/apache_beam/typehints/schemas_test.py
index 5c202a3aa3d..a41df70c5a8 100644
--- a/sdks/python/apache_beam/typehints/schemas_test.py
+++ b/sdks/python/apache_beam/typehints/schemas_test.py
@@ -25,14 +25,11 @@ import datetime
import itertools
import pickle
import unittest
+from collections.abc import Mapping
+from collections.abc import Sequence
from typing import Any
-from typing import ByteString
-from typing import List
-from typing import Mapping
from typing import NamedTuple
from typing import Optional
-from typing import Sequence
-from typing import Tuple
import numpy as np
import pytest
@@ -57,6 +54,12 @@ from apache_beam.typehints.schemas import
typing_to_runner_api
from apache_beam.typehints.testing.strategies import named_fields
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
+
all_nonoptional_primitives = [
np.int8,
np.int16,
@@ -81,8 +84,8 @@ basic_map_types = [
]
basic_tuple_types = [
- Tuple[str, np.int64],
- Tuple[np.int64, ...],
+ tuple[str, np.int64],
+ tuple[np.int64, ...],
]
@@ -115,8 +118,8 @@ class ComplexSchema(NamedTuple):
array_optional: Sequence[Optional[bool]]
timestamp: Timestamp
date: datetime.date
- fixed_tuple: Tuple[str, np.int64]
- var_tuple: Tuple[np.int64, ...]
+ fixed_tuple: tuple[str, np.int64]
+ var_tuple: tuple[np.int64, ...]
def get_test_beam_fieldtype_protos():
@@ -667,9 +670,9 @@ class SchemaTest(unittest.TestCase):
[
('name', str),
('age', Optional[int]),
- ('interests', List[str]),
+ ('interests', list[str]),
('height', float),
- ('blob', ByteString),
+ ('blob', Buffer),
])
expected = schema_pb2.FieldType(
@@ -712,9 +715,9 @@ class SchemaTest(unittest.TestCase):
class MyCuteDataclass:
name: str
age: Optional[int]
- interests: List[str]
+ interests: list[str]
height: float
- blob: ByteString
+ blob: Buffer
expected = schema_pb2.FieldType(
row_type=schema_pb2.RowType(
diff --git a/sdks/python/apache_beam/typehints/testing/strategies.py
b/sdks/python/apache_beam/typehints/testing/strategies.py
index a388e25eea2..97afc16d737 100644
--- a/sdks/python/apache_beam/typehints/testing/strategies.py
+++ b/sdks/python/apache_beam/typehints/testing/strategies.py
@@ -21,9 +21,9 @@ Intended for internal use only, no backward-compatibility
guarantees."""
import keyword
import unicodedata
-from typing import Mapping
+from collections.abc import Mapping
+from collections.abc import Sequence
from typing import Optional
-from typing import Sequence
from hypothesis import strategies as st
