You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2025/03/05 19:58:36 UTC
(beam) branch master updated: Correctly parse labels if they are passed as a single string instead of a list (#34183)
This is an automated email from the ASF dual-hosted git repository.
damccorm 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 9da70060e8f Correctly parse labels if they are passed as a single string instead of a list (#34183)
9da70060e8f is described below
commit 9da70060e8f9cd081a669c73daa6b7b82ccef170
Author: Danny McCormick <da...@google.com>
AuthorDate: Wed Mar 5 14:58:30 2025 -0500
Correctly parse labels if they are passed as a single string instead of a list (#34183)
* Correctly parse labels if they are passed as a single string instead of a list
* Fix dict case adnd add tests
* format
---
.../runners/dataflow/internal/apiclient.py | 4 +++
.../runners/dataflow/internal/apiclient_test.py | 40 ++++++++++++++++++++++
2 files changed, 44 insertions(+)
diff --git a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
index 4e65156f3bc..a757d45c036 100644
--- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
+++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
@@ -447,6 +447,10 @@ class Job(object):
if self.google_cloud_options.labels:
self.proto.labels = dataflow.Job.LabelsValue()
labels = self.google_cloud_options.labels
+ if isinstance(labels, str):
+ labels = [labels]
+ elif isinstance(labels, dict):
+ labels = [str(labels)]
for label in labels:
if '{' in label:
label = ast.literal_eval(label)
diff --git a/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py b/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
index d055065cb9d..68a2617bbaf 100644
--- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
+++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
@@ -885,6 +885,46 @@ class UtilTest(unittest.TestCase):
self.assertEqual('count', job.proto.labels.additionalProperties[2].key)
self.assertEqual('3', job.proto.labels.additionalProperties[2].value)
+ pipeline_options = PipelineOptions([
+ '--project',
+ 'test_project',
+ '--job_name',
+ 'test_job_name',
+ '--temp_location',
+ 'gs://test-location/temp'
+ ])
+ pipeline_options.view_as(GoogleCloudOptions).labels = {
+ "name": "wrench", "mass": "1_3kg", "count": "3"
+ }
+ job = apiclient.Job(pipeline_options, beam_runner_api_pb2.Pipeline())
+ self.assertEqual(3, len(job.proto.labels.additionalProperties))
+ self.assertEqual('name', job.proto.labels.additionalProperties[0].key)
+ self.assertEqual('wrench', job.proto.labels.additionalProperties[0].value)
+ self.assertEqual('mass', job.proto.labels.additionalProperties[1].key)
+ self.assertEqual('1_3kg', job.proto.labels.additionalProperties[1].value)
+ self.assertEqual('count', job.proto.labels.additionalProperties[2].key)
+ self.assertEqual('3', job.proto.labels.additionalProperties[2].value)
+
+ pipeline_options = PipelineOptions([
+ '--project',
+ 'test_project',
+ '--job_name',
+ 'test_job_name',
+ '--temp_location',
+ 'gs://test-location/temp'
+ ])
+ pipeline_options.view_as(
+ GoogleCloudOptions
+ ).labels = '{ "name": "wrench", "mass": "1_3kg", "count": "3" }'
+ job = apiclient.Job(pipeline_options, beam_runner_api_pb2.Pipeline())
+ self.assertEqual(3, len(job.proto.labels.additionalProperties))
+ self.assertEqual('name', job.proto.labels.additionalProperties[0].key)
+ self.assertEqual('wrench', job.proto.labels.additionalProperties[0].value)
+ self.assertEqual('mass', job.proto.labels.additionalProperties[1].key)
+ self.assertEqual('1_3kg', job.proto.labels.additionalProperties[1].value)
+ self.assertEqual('count', job.proto.labels.additionalProperties[2].key)
+ self.assertEqual('3', job.proto.labels.additionalProperties[2].value)
+
def test_experiment_use_multiple_sdk_containers(self):
pipeline_options = PipelineOptions([
'--project',