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 fc0d9895080 Replace non-PEP 585 types in watch.py (#39527)
fc0d9895080 is described below
commit fc0d989508095c669e153e7d0c0327bbe3349cf5
Author: Jack McCluskey <[email protected]>
AuthorDate: Tue Jul 28 13:28:10 2026 -0400
Replace non-PEP 585 types in watch.py (#39527)
* Fix legacy typing.Tuple usage in watch.py
* Swap to the correct iterable
---
sdks/python/apache_beam/io/watch.py | 11 +++++------
1 file changed, 5 insertions(+), 6 deletions(-)
diff --git a/sdks/python/apache_beam/io/watch.py
b/sdks/python/apache_beam/io/watch.py
index 2fcee7a8080..b0ff2bd6a07 100644
--- a/sdks/python/apache_beam/io/watch.py
+++ b/sdks/python/apache_beam/io/watch.py
@@ -59,12 +59,11 @@ import hashlib
import inspect
import time
import typing
+from collections.abc import Iterable
from typing import Any
from typing import Callable
from typing import Generic
-from typing import Iterable
from typing import Optional
-from typing import Tuple
from typing import TypeVar
from apache_beam import coders
@@ -111,7 +110,7 @@ class PollResult(Generic[OutputT]):
The ``OutputT`` type parameter can annotate a poll function's return type,
as in ``-> PollResult[str]``; the transform infers the output coder from it.
"""
- outputs: Tuple[TimestampedValue, ...]
+ outputs: tuple[TimestampedValue, ...]
watermark: Optional[Timestamp] = None
@property
@@ -119,7 +118,7 @@ class PollResult(Generic[OutputT]):
return self.watermark == MAX_TIMESTAMP
@staticmethod
- def _normalize(outputs, timestamp) -> Tuple[TimestampedValue, ...]:
+ def _normalize(outputs, timestamp) -> tuple[TimestampedValue, ...]:
if timestamp is None:
default_ts = Timestamp.now()
else:
@@ -428,7 +427,7 @@ class _GrowthRestrictionTracker(iobase.RestrictionTracker):
def current_restriction(self) -> _GrowthState:
return self._restriction
- def try_claim(self, position: Tuple[PollResult, Any]) -> bool:
+ def try_claim(self, position: tuple[PollResult, Any]) -> bool:
"""Claims one poll round; at most one claim succeeds per ``process()``.
The claim is rejected after a checkpoint already stopped this invocation,
@@ -723,7 +722,7 @@ class Watch(PTransform):
output_coder,
key_fn,
key_coder,
- self._now)).with_output_types(Tuple[input_type, value_type])
+ self._now)).with_output_types(tuple[input_type, value_type])
def _as_duration(value) -> Duration:
