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 da7693c3dd1 [Gemini] Fix pyrefly check invalid-yield (#39417)
da7693c3dd1 is described below
commit da7693c3dd17679d1e9493a33e7e526053365ce4
Author: Jack McCluskey <[email protected]>
AuthorDate: Wed Jul 22 13:07:32 2026 -0400
[Gemini] Fix pyrefly check invalid-yield (#39417)
* Fix pyrefly check invalid-yield
* yapf
---
.../inference/pytorch_image_classification_with_side_inputs.py | 2 +-
sdks/python/apache_beam/io/gcp/bigquery_change_history.py | 10 +++++++---
sdks/python/pyproject.toml | 1 -
3 files changed, 8 insertions(+), 5 deletions(-)
diff --git
a/sdks/python/apache_beam/examples/inference/pytorch_image_classification_with_side_inputs.py
b/sdks/python/apache_beam/examples/inference/pytorch_image_classification_with_side_inputs.py
index 787341263fd..903bf38ab27 100644
---
a/sdks/python/apache_beam/examples/inference/pytorch_image_classification_with_side_inputs.py
+++
b/sdks/python/apache_beam/examples/inference/pytorch_image_classification_with_side_inputs.py
@@ -115,7 +115,7 @@ class PostProcessor(beam.DoFn):
Return filename, prediction and the model id used to perform the
prediction
"""
- def process(self, element: tuple[str, PredictionResult]) -> Iterable[str]:
+ def process(self, element: tuple[str, PredictionResult]) -> Iterable[Any]:
filename, prediction_result = element
prediction = torch.argmax(prediction_result.inference, dim=0)
yield filename, prediction, prediction_result.model_id
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_change_history.py
b/sdks/python/apache_beam/io/gcp/bigquery_change_history.py
index f0a23ddce02..1e90c297477 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_change_history.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery_change_history.py
@@ -496,7 +496,8 @@ class _PollChangeHistoryFn(beam.DoFn,
beam.transforms.core.RestrictionProvider):
self,
start_ts: Timestamp,
end_ts: Timestamp,
- watermark_estimator: _PollWatermarkEstimator) -> Iterable[_QueryRange]:
+ watermark_estimator: _PollWatermarkEstimator
+ ) -> Iterable[TimestampedValue[_QueryRange]]:
"""Compute and yield _QueryRange elements, advancing estimator state."""
ranges = compute_ranges(start_ts, end_ts, self._change_function)
_LOGGER.info(
@@ -524,7 +525,7 @@ class _PollChangeHistoryFn(beam.DoFn,
beam.transforms.core.RestrictionProvider):
restriction_tracker=beam.DoFn.RestrictionParam(),
watermark_estimator=beam.DoFn.WatermarkEstimatorParam(
_PollWatermarkEstimatorProvider())
- ) -> Iterable[_QueryRange]:
+ ) -> Iterable[TimestampedValue[_QueryRange]]:
now = time.time()
start_ts = watermark_estimator.poll_cursor()
@@ -1031,7 +1032,10 @@ class _DecompressArrowBatchesFn(beam.DoFn):
def __init__(self, change_timestamp_column: str = 'change_timestamp') ->
None:
self._change_timestamp_column = change_timestamp_column
- def process(self, element: tuple[bytes, bytes]) -> Iterable[dict[str, Any]]:
+ def process(
+ self,
+ element: tuple[bytes,
+ bytes]) -> Iterable[TimestampedValue[dict[str, Any]]]:
schema_bytes, batch_bytes = element
schema = pyarrow.ipc.read_schema(pyarrow.py_buffer(schema_bytes))
batch = pyarrow.ipc.read_record_batch(
diff --git a/sdks/python/pyproject.toml b/sdks/python/pyproject.toml
index e769bf2d91e..5901d817c9c 100644
--- a/sdks/python/pyproject.toml
+++ b/sdks/python/pyproject.toml
@@ -212,5 +212,4 @@ invalid-argument = "ignore"
invalid-inheritance = "ignore"
not-iterable = "ignore"
unexpected-keyword = "ignore"
-invalid-yield = "ignore"
bad-typed-dict-key = "ignore"
