Fix Dataflow ValueProvider serialization

CHANGES.md

@@ -112,6 +112,7 @@
 
 ## Bugfixes
 
+* Fixed unresolved runtime `ValueProvider` options being stringified in Python Dataflow Flex Templates ([#39499](https://github.com/apache/beam/issues/39499)).
 * Fixed unbounded checkpoint state growth for splittable DoFns that self-checkpoint on the portable Flink runner (Java) ([#27648](https://github.com/apache/beam/issues/27648)).
 * Improved Java pipeline performance by avoiding repeated `DoFn` type descriptor resolution when creating cached invokers ([#39309](https://github.com/apache/beam/issues/39309)).
 * (Python) Fixed a memory leak in Python SDK caused by storing exceptions with potentially large stack frames in a cache ([#39406](https://github.com/apache/beam/issues/39406)).

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

@@ -280,8 +280,10 @@ def __init__(
       for k, v in sdk_pipeline_options.items():
         if v is None:
           continue
-        options_dict[k] = str(v) if isinstance(
-            v, value_provider.ValueProvider) else v
+        if isinstance(v, value_provider.ValueProvider):
+          options_dict[k] = v.get() if v.is_accessible() else None
+        else:
+          options_dict[k] = v
       options_dict["pipelineUrl"] = proto_pipeline_staged_url
       if pipeline_proto_hash:
         options_dict["pipelineProtoHash"] = pipeline_proto_hash

sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py

@@ -113,6 +113,25 @@ def test_pipeline_url(self):
 
     self.assertEqual(pipeline_url, FAKE_PIPELINE_URL)
 
+  def test_value_provider_options_serialization(self):
+    class UserOptions(PipelineOptions):
+      @classmethod
+      def _add_argparse_args(cls, parser):
+        parser.add_value_provider_argument('--at_vp_arg1')
+        parser.add_value_provider_argument('--at_vp_arg2')
+
+    pipeline_options = UserOptions([
+        '--at_vp_arg2', 'provided', '--temp_location', 'gs://any-location/temp'
+    ])
+    env = apiclient.Environment([],
+                                pipeline_options,
+                                '2.0.0',
+                                FAKE_PIPELINE_URL)
+
+    recovered_options = env.proto.sdk_pipeline_options['options']
+    self.assertIsNone(recovered_options['at_vp_arg1'])
+    self.assertEqual(recovered_options['at_vp_arg2'], 'provided')
+
   def test_pipeline_proto_hash(self):
     pipeline_options = PipelineOptions(
         ['--temp_location', 'gs://any-location/temp'])