Use add_experiment() instead of experiments.append() in Python

Replaces manual list manipulation pattern with DebugOptions.add_experiment()
which handles null-init and deduplication internally.

Resolves #19347

Author: ash6898

sdks/python/apache_beam/io/external/xlang_parquetio_test.py

@@ -51,8 +51,7 @@ def test_xlang_parquetio_write(self):
     address = 'localhost:%s' % port
     try:
       with TestPipeline() as p:
-        p.get_pipeline_options().view_as(DebugOptions).experiments.append(
-            'jar_packages=' + expansion_jar)
+        p.get_pipeline_options().view_as(DebugOptions).add_experiment('jar_packages=' + expansion_jar)
         p.not_use_test_runner_api = True
         _ = p \
           | beam.Create([

sdks/python/apache_beam/io/iobase_test.py

@@ -196,14 +196,7 @@ def test_try_split_with_any_exception(self):
 class UseSdfBoundedSourcesTests(unittest.TestCase):
   def _run_sdf_wrapper_pipeline(self, source, expected_values):
     with beam.Pipeline() as p:
-      experiments = (p._options.view_as(DebugOptions).experiments or [])
-
-      # Setup experiment option to enable using SDFBoundedSourceWrapper
-      if 'beam_fn_api' not in experiments:
-        # Required so mocking below doesn't mock Create used in assert_that.
-        experiments.append('beam_fn_api')
-
-      p._options.view_as(DebugOptions).experiments = experiments
+      p._options.view_as(DebugOptions).add_experiment('beam_fn_api')
 
       actual = p | beam.io.Read(source)
       assert_that(actual, equal_to(expected_values))

sdks/python/apache_beam/pipeline.py

@@ -232,10 +232,7 @@ def __init__(
     # set default experiments for portable runners
     # (needs to occur prior to pipeline construction)
     if runner.is_fnapi_compatible():
-      experiments = (self._options.view_as(DebugOptions).experiments or [])
-      if not 'beam_fn_api' in experiments:
-        experiments.append('beam_fn_api')
-        self._options.view_as(DebugOptions).experiments = experiments
+      self._options.view_as(DebugOptions).add_experiment('beam_fn_api')
 
     self.local_tempdir = tempfile.mkdtemp(prefix='beam-pipeline-temp')
 

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

@@ -147,16 +147,12 @@ def __init__(
     ])
     # TODO: Use enumerated type instead of strings for job types.
     if job_type.startswith('FNAPI_'):
-      self.debug_options.experiments = self.debug_options.experiments or []
-
-      debug_options_experiments = self.debug_options.experiments
       # Add use_multiple_sdk_containers flag if it's not already present. Do not
       # add the flag if 'no_use_multiple_sdk_containers' is present.
       # TODO: Cleanup use_multiple_sdk_containers once we deprecate Python SDK
       # till version 2.4.
-      if ('use_multiple_sdk_containers' not in debug_options_experiments and
-          'no_use_multiple_sdk_containers' not in debug_options_experiments):
-        debug_options_experiments.append('use_multiple_sdk_containers')
+      if ('no_use_multiple_sdk_containers' not in (self.debug_options.experiments or [])):
+        self.debug_options.add_experiment('use_multiple_sdk_containers')
     # FlexRS
     if self.google_cloud_options.flexrs_goal == 'COST_OPTIMIZED':
       self.proto.flexResourceSchedulingGoal = (

sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py

@@ -141,11 +141,7 @@ def run_pipeline(
     RuntimeValueProvider.set_runtime_options({})
 
     # Setup "beam_fn_api" experiment options if lacked.
-    experiments = (
-        options.view_as(pipeline_options.DebugOptions).experiments or [])
-    if not 'beam_fn_api' in experiments:
-      experiments.append('beam_fn_api')
-    options.view_as(pipeline_options.DebugOptions).experiments = experiments
+    options.view_as(pipeline_options.DebugOptions).add_experiment('beam_fn_api')
 
     # This is sometimes needed if type checking is disabled
     # to enforce that the inputs (and outputs) of GroupByKey operations