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 9b5600d7156d..aa14a1787453 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 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 70293f863f47..f85b77e7a298 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 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) 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 d3d608b1fc6f..bd523835daaa 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 @@ """ 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 73cedb3a6aee..788f872ced65 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 @@ 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.""" diff --git a/sdks/python/apache_beam/typehints/row_type.py b/sdks/python/apache_beam/typehints/row_type.py index 5e0faf91f55b..89e15b5df822 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 @@ 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 diff --git a/sdks/python/apache_beam/typehints/schemas.py b/sdks/python/apache_beam/typehints/schemas.py index 71bd540c522a..11cea0195436 100644 --- a/sdks/python/apache_beam/typehints/schemas.py +++ b/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,7 +242,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) @@ -248,7 +250,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) @@ -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] diff --git a/sdks/python/apache_beam/typehints/schemas_test.py b/sdks/python/apache_beam/typehints/schemas_test.py index 5c202a3aa3df..a41df70c5a8e 100644 --- a/sdks/python/apache_beam/typehints/schemas_test.py +++ b/sdks/python/apache_beam/typehints/schemas_test.py @@ -25,14 +25,11 @@ 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.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_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 @@ def test_trivial_example(self): [ ('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 @@ def test_trivial_example_dataclass(self): 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 a388e25eea25..97afc16d7378 100644 --- a/sdks/python/apache_beam/typehints/testing/strategies.py +++ b/sdks/python/apache_beam/typehints/testing/strategies.py @@ -21,9 +21,9 @@ 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