From 058a4d7003404f5923f6dbd4bd456d1e5ad8cbce Mon Sep 17 00:00:00 2001 From: jrmccluskey Date: Tue, 21 Jul 2026 19:38:30 +0000 Subject: [PATCH 1/2] Fix pyrefly check invalid-yield --- .../pytorch_image_classification_with_side_inputs.py | 4 +++- .../python/apache_beam/io/gcp/bigquery_change_history.py | 9 ++++++--- sdks/python/pyproject.toml | 1 - 3 files changed, 9 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 787341263fde..3efb1658907c 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,9 @@ 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 f0a23ddce02a..aa51ae6797bd 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 @@ def _emit_query_ranges( 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 @@ def process( 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,9 @@ 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 52f713466246..6dea71438057 100644 --- a/sdks/python/pyproject.toml +++ b/sdks/python/pyproject.toml @@ -214,6 +214,5 @@ not-iterable = "ignore" unexpected-keyword = "ignore" bad-specialization = "ignore" bad-context-manager = "ignore" -invalid-yield = "ignore" bad-argument-count = "ignore" bad-typed-dict-key = "ignore" From 56f53e59d5340c6f8b9368ce7a8da48334c38e5c Mon Sep 17 00:00:00 2001 From: jrmccluskey Date: Wed, 22 Jul 2026 13:07:14 +0000 Subject: [PATCH 2/2] yapf --- .../pytorch_image_classification_with_side_inputs.py | 4 +--- sdks/python/apache_beam/io/gcp/bigquery_change_history.py | 5 +++-- 2 files changed, 4 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 3efb1658907c..903bf38ab279 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,9 +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[Any]: + 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 aa51ae6797bd..1e90c2974778 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery_change_history.py +++ b/sdks/python/apache_beam/io/gcp/bigquery_change_history.py @@ -1033,8 +1033,9 @@ 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[TimestampedValue[dict[str, Any]]]: + 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(