From 8af0efc15fbcc671467a87d18e3041ad635a4205 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Mon, 12 Oct 2026 10:07:12 -0400 Subject: [PATCH 1/2] [Python] Add AwsOptions and deprecate S3Options --- .../io/aws/clients/s3/boto3_client.py | 29 ++--- .../io/aws/clients/s3/boto3_client_test.py | 111 ++++++++++++++++++ .../io/aws/clients/s3/client_test.py | 2 +- .../python/apache_beam/io/aws/s3filesystem.py | 2 +- sdks/python/apache_beam/io/aws/s3io_test.py | 2 +- .../apache_beam/options/pipeline_options.py | 106 ++++++++++++++--- .../options/pipeline_options_test.py | 54 +++++++++ 7 files changed, 268 insertions(+), 38 deletions(-) create mode 100644 sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py diff --git a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py index aee24ac2c052..9182c1d7df80 100644 --- a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py +++ b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py @@ -43,24 +43,19 @@ class Client(object): def __init__(self, options): assert boto3 is not None, 'Missing boto3 requirement' if isinstance(options, pipeline_options.PipelineOptions): - s3_options = options.view_as(pipeline_options.S3Options) - access_key_id = s3_options.s3_access_key_id - secret_access_key = s3_options.s3_secret_access_key - session_token = s3_options.s3_session_token - endpoint_url = s3_options.s3_endpoint_url - use_ssl = not s3_options.s3_disable_ssl - region_name = s3_options.s3_region_name - api_version = s3_options.s3_api_version - verify = s3_options.s3_verify + aws_options = options.view_as(pipeline_options.AwsOptions) else: - access_key_id = options.get('s3_access_key_id') - secret_access_key = options.get('s3_secret_access_key') - session_token = options.get('s3_session_token') - endpoint_url = options.get('s3_endpoint_url') - use_ssl = not options.get('s3_disable_ssl', False) - region_name = options.get('s3_region_name') - api_version = options.get('s3_api_version') - verify = options.get('s3_verify') + # A plain dict of option names (aws_* or the deprecated s3_*) to values. + aws_options = pipeline_options.AwsOptions(flags=[], **options) + + access_key_id = aws_options.aws_access_key_id + secret_access_key = aws_options.aws_secret_access_key + session_token = aws_options.aws_session_token + endpoint_url = aws_options.aws_endpoint_url + use_ssl = not aws_options.aws_disable_ssl + region_name = aws_options.aws_region_name + api_version = aws_options.aws_api_version + verify = aws_options.aws_verify session = boto3.session.Session() self.client = session.client( diff --git a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py new file mode 100644 index 000000000000..ccea8a3c6640 --- /dev/null +++ b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py @@ -0,0 +1,111 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Tests for option handling in the boto3 S3 client.""" + +import unittest +from unittest import mock + +from apache_beam.options.pipeline_options import PipelineOptions + +try: + import boto3 # pylint: disable=unused-import + + from apache_beam.io.aws.clients.s3 import boto3_client +except ImportError: + boto3_client = None + + +@unittest.skipIf(boto3_client is None, 'boto3 is not installed') +class Boto3ClientOptionsTest(unittest.TestCase): + def setUp(self): + patcher = mock.patch('boto3.session.Session') + self.mock_session = patcher.start().return_value + self.addCleanup(patcher.stop) + + def _client_kwargs(self): + return self.mock_session.client.call_args.kwargs + + def test_aws_options_passed_to_client(self): + boto3_client.Client( + PipelineOptions([ + '--aws_access_key_id=AKIDEXAMPLE', + '--aws_secret_access_key=SECRETEXAMPLE', + '--aws_session_token=TOKEN', + '--aws_endpoint_url=http://localhost:4566', + '--aws_region_name=us-east-1', + '--aws_api_version=2006-03-01', + '--aws_verify=False', + '--aws_disable_ssl', + ])) + self.assertEqual( + self._client_kwargs(), + { + 'service_name': 's3', + 'region_name': 'us-east-1', + 'api_version': '2006-03-01', + 'use_ssl': False, + 'verify': 'False', + 'endpoint_url': 'http://localhost:4566', + 'aws_access_key_id': 'AKIDEXAMPLE', + 'aws_secret_access_key': 'SECRETEXAMPLE', + 'aws_session_token': 'TOKEN', + }) + + def test_deprecated_s3_options_still_work(self): + boto3_client.Client( + PipelineOptions([ + '--s3_access_key_id=AKIDEXAMPLE', + '--s3_secret_access_key=SECRETEXAMPLE', + '--s3_region_name=us-west-2', + '--s3_disable_ssl', + ])) + kwargs = self._client_kwargs() + self.assertEqual(kwargs['aws_access_key_id'], 'AKIDEXAMPLE') + self.assertEqual(kwargs['aws_secret_access_key'], 'SECRETEXAMPLE') + self.assertEqual(kwargs['region_name'], 'us-west-2') + self.assertFalse(kwargs['use_ssl']) + + def test_aws_options_take_precedence_over_s3_options(self): + boto3_client.Client( + PipelineOptions([ + '--s3_region_name=us-west-2', + '--aws_region_name=us-east-1', + ])) + self.assertEqual(self._client_kwargs()['region_name'], 'us-east-1') + + def test_dict_options(self): + boto3_client.Client({ + 'aws_access_key_id': 'AKIDEXAMPLE', + 's3_secret_access_key': 'SECRETEXAMPLE', + 'aws_disable_ssl': True, + }) + kwargs = self._client_kwargs() + self.assertEqual(kwargs['aws_access_key_id'], 'AKIDEXAMPLE') + self.assertEqual(kwargs['aws_secret_access_key'], 'SECRETEXAMPLE') + self.assertFalse(kwargs['use_ssl']) + + def test_no_options_uses_boto3_defaults(self): + boto3_client.Client(PipelineOptions([])) + kwargs = self._client_kwargs() + self.assertIsNone(kwargs['aws_access_key_id']) + self.assertIsNone(kwargs['aws_secret_access_key']) + self.assertTrue(kwargs['use_ssl']) + + +if __name__ == '__main__': + unittest.main() diff --git a/sdks/python/apache_beam/io/aws/clients/s3/client_test.py b/sdks/python/apache_beam/io/aws/clients/s3/client_test.py index 67797b1736b9..ea110ea9b5d8 100644 --- a/sdks/python/apache_beam/io/aws/clients/s3/client_test.py +++ b/sdks/python/apache_beam/io/aws/clients/s3/client_test.py @@ -47,7 +47,7 @@ def setUp(self): self.client.known_buckets.add(test_data_bucket) self.aws = s3io.S3IO(self.client) else: - self.aws = s3io.S3IO(options=pipeline_options.S3Options()) + self.aws = s3io.S3IO(options=pipeline_options.AwsOptions()) def test_get_object_metadata(self): diff --git a/sdks/python/apache_beam/io/aws/s3filesystem.py b/sdks/python/apache_beam/io/aws/s3filesystem.py index 584263ec241e..f7ec8a28d2b0 100644 --- a/sdks/python/apache_beam/io/aws/s3filesystem.py +++ b/sdks/python/apache_beam/io/aws/s3filesystem.py @@ -41,7 +41,7 @@ def __init__(self, pipeline_options): """Initializes a connection to S3. Connection configuration is done by passing pipeline options. - See :class:`~apache_beam.options.pipeline_options.S3Options`. + See :class:`~apache_beam.options.pipeline_options.AwsOptions`. """ super().__init__(pipeline_options) self._options = pipeline_options diff --git a/sdks/python/apache_beam/io/aws/s3io_test.py b/sdks/python/apache_beam/io/aws/s3io_test.py index 09df3da7674f..ee871be9d823 100644 --- a/sdks/python/apache_beam/io/aws/s3io_test.py +++ b/sdks/python/apache_beam/io/aws/s3io_test.py @@ -98,7 +98,7 @@ def setUp(self): self.aws = s3io.S3IO(self.client) else: - self.aws = s3io.S3IO(options=pipeline_options.S3Options()) + self.aws = s3io.S3IO(options=pipeline_options.AwsOptions()) self.client = self.aws.client def test_size(self): diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py index 743ab967b04c..37a98b638209 100644 --- a/sdks/python/apache_beam/options/pipeline_options.py +++ b/sdks/python/apache_beam/options/pipeline_options.py @@ -51,6 +51,7 @@ 'ProfilingOptions', 'SetupOptions', 'TestOptions', + 'AwsOptions', 'S3Options' ] @@ -2398,42 +2399,111 @@ def augment_options(cls, options): return options -class S3Options(PipelineOptions): +# Deprecated S3Options option names, mapped to the AwsOptions options that +# replace them. +_DEPRECATED_S3_TO_AWS_OPTIONS = { + 's3_access_key_id': 'aws_access_key_id', + 's3_secret_access_key': 'aws_secret_access_key', + 's3_session_token': 'aws_session_token', + 's3_endpoint_url': 'aws_endpoint_url', + 's3_region_name': 'aws_region_name', + 's3_api_version': 'aws_api_version', + 's3_verify': 'aws_verify', + 's3_disable_ssl': 'aws_disable_ssl', +} +_AWS_TO_DEPRECATED_S3_OPTIONS = { + aws: s3 + for s3, aws in _DEPRECATED_S3_TO_AWS_OPTIONS.items() +} + + +class AwsOptions(PipelineOptions): + """Options used to create AWS service clients, e.g. for S3. + + The deprecated ``--s3_*`` flags from :class:`S3Options` are still accepted. + Each ``aws_*`` option falls back to its ``s3_*`` counterpart when it is not + set. + """ @classmethod def _add_argparse_args(cls, parser): - # These options are passed to the S3 IO Client parser.add_argument( - '--s3_access_key_id', + '--aws_access_key_id', default=None, - help='The secret key to use when creating the s3 client.') + help='The access key ID to use when creating AWS clients.') parser.add_argument( - '--s3_secret_access_key', + '--aws_secret_access_key', default=None, - help='The secret key to use when creating the s3 client.') + help='The secret access key to use when creating AWS clients.') parser.add_argument( - '--s3_session_token', + '--aws_session_token', default=None, - help='The session token to use when creating the s3 client.') + help='The session token to use when creating AWS clients.') parser.add_argument( - '--s3_endpoint_url', + '--aws_endpoint_url', default=None, - help='The complete URL to use for the constructed s3 client.') + help='The complete URL to use for the constructed AWS client.') parser.add_argument( - '--s3_region_name', + '--aws_region_name', default=None, - help='The name of the region associated with the s3 client.') + help='The name of the region associated with the AWS client.') parser.add_argument( - '--s3_api_version', + '--aws_api_version', default=None, - help='The API version to use with the s3 client.') + help='The API version to use with the AWS client.') parser.add_argument( - '--s3_verify', + '--aws_verify', default=None, - help='Whether or not to verify SSL certificates with the s3 client.') + help='Whether or not to verify SSL certificates with the AWS client.') parser.add_argument( - '--s3_disable_ssl', + '--aws_disable_ssl', default=False, action='store_true', help=( - 'Whether or not to use SSL with the s3 client. ' + 'Whether or not to use SSL with the AWS client. ' 'By default, SSL is used.')) + + # Deprecated S3Options flags, kept for backward compatibility. They are + # registered here (not on S3Options) so that both flag sets are parsed by + # every AwsOptions view and are never registered twice. + for s3_name, aws_name in _DEPRECATED_S3_TO_AWS_OPTIONS.items(): + if s3_name == 's3_disable_ssl': + kwargs = {'default': False, 'action': 'store_true'} + else: + kwargs = {'default': None} + parser.add_argument( + '--' + s3_name, + help='Deprecated, use --%s instead.' % aws_name, + **kwargs) + + def __getattr__(self, name): + value = super().__getattr__(name) + s3_name = _AWS_TO_DEPRECATED_S3_OPTIONS.get(name) + if s3_name and not value: + s3_value = self._all_options.get(s3_name) + if s3_value: + logger.log_first_n( + logging.WARN, + '--%s is deprecated, use --%s instead.' % (s3_name, name), + key='message') + return s3_value + return value + + +class S3Options(AwsOptions): + """Deprecated, use :class:`AwsOptions` instead. + + The ``s3_*`` attributes delegate to the corresponding ``aws_*`` options, so + both the old and new flags are honored. + """ + def __init__(self, flags: Optional[Sequence[str]] = None, **kwargs) -> None: + logger.log_first_n( + logging.WARN, + 'S3Options is deprecated, use AwsOptions instead.', + key='message') + super().__init__(flags, **kwargs) + + def __getattr__(self, name): + aws_name = _DEPRECATED_S3_TO_AWS_OPTIONS.get(name) + if aws_name: + return getattr(self, aws_name) + return super().__getattr__(name) diff --git a/sdks/python/apache_beam/options/pipeline_options_test.py b/sdks/python/apache_beam/options/pipeline_options_test.py index d9e1d850ec7f..f02ec10a69a4 100644 --- a/sdks/python/apache_beam/options/pipeline_options_test.py +++ b/sdks/python/apache_beam/options/pipeline_options_test.py @@ -28,12 +28,14 @@ import mock from parameterized import parameterized +from apache_beam.options.pipeline_options import AwsOptions from apache_beam.options.pipeline_options import CrossLanguageOptions from apache_beam.options.pipeline_options import DebugOptions from apache_beam.options.pipeline_options import GoogleCloudOptions from apache_beam.options.pipeline_options import JobServerOptions from apache_beam.options.pipeline_options import PipelineOptions from apache_beam.options.pipeline_options import ProfilingOptions +from apache_beam.options.pipeline_options import S3Options from apache_beam.options.pipeline_options import StandardOptions from apache_beam.options.pipeline_options import TypeOptions from apache_beam.options.pipeline_options import WorkerOptions @@ -1121,6 +1123,58 @@ def test_is_compat_version_prior_to(self): msg="Should return False when update_compatibility_version is not set") +class AwsOptionsTest(unittest.TestCase): + def test_aws_options(self): + options = PipelineOptions([ + '--aws_access_key_id=key', + '--aws_region_name=us-east-1', + '--aws_disable_ssl', + ]).view_as(AwsOptions) + self.assertEqual(options.aws_access_key_id, 'key') + self.assertEqual(options.aws_region_name, 'us-east-1') + self.assertTrue(options.aws_disable_ssl) + self.assertIsNone(options.aws_endpoint_url) + + def test_aws_options_fall_back_to_deprecated_s3_flags(self): + options = PipelineOptions([ + '--s3_access_key_id=key', + '--s3_region_name=us-west-2', + '--s3_disable_ssl', + ]).view_as(AwsOptions) + self.assertEqual(options.aws_access_key_id, 'key') + self.assertEqual(options.aws_region_name, 'us-west-2') + self.assertTrue(options.aws_disable_ssl) + + def test_aws_flags_take_precedence_over_s3_flags(self): + options = PipelineOptions([ + '--s3_region_name=us-west-2', + '--aws_region_name=us-east-1', + ]).view_as(AwsOptions) + self.assertEqual(options.aws_region_name, 'us-east-1') + + def test_aws_options_fall_back_to_deprecated_s3_kwargs(self): + options = PipelineOptions(s3_endpoint_url='http://localhost:4566') + self.assertEqual( + options.view_as(AwsOptions).aws_endpoint_url, 'http://localhost:4566') + + def test_s3_options_delegate_to_aws_options(self): + options = PipelineOptions([ + '--aws_access_key_id=key', + '--s3_region_name=us-west-2', + ]).view_as(S3Options) + self.assertIsInstance(options, AwsOptions) + self.assertEqual(options.s3_access_key_id, 'key') + self.assertEqual(options.s3_region_name, 'us-west-2') + self.assertEqual(options.aws_region_name, 'us-west-2') + self.assertFalse(options.s3_disable_ssl) + + def test_options_are_registered_once(self): + all_options = PipelineOptions(['--aws_region_name=us-east-1' + ]).get_all_options() + self.assertEqual(all_options['aws_region_name'], 'us-east-1') + self.assertIn('s3_region_name', all_options) + + if __name__ == '__main__': logging.getLogger().setLevel(logging.INFO) unittest.main() From de09a520e11c55d02e448a12f96afeeb235a7d89 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Mon, 12 Oct 2026 10:14:57 -0400 Subject: [PATCH 2/2] Include the changes in CHANGES.md --- CHANGES.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 6b01d163823c..cb0059c3cfb2 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -73,6 +73,7 @@ * The `disableCounterMetrics`, `disableStringSetMetrics` and `disableBoundedTrieMetrics` experiments are now honored by the Python SDK, as they already were in Java (Python) ([#38746](https://github.com/apache/beam/issues/38746)). * (Go) Pipeline-level `--resource_hints` are now forwarded to expansion services, so they apply to cross-language transforms as they do in the Java and Python SDKs. The `max_active_bundles_per_worker` hint name is now also accepted by `--resource_hints` ([#23893](https://github.com/apache/beam/issues/23893)). * ReadFromBigQuery now supports Lakehouse runtime catalog (BigLake metastore) tables with `method=DIRECT_READ`, using 4-part `project.catalog.namespace.table` identifiers. Previously `project:catalog.namespace.table` was silently mis-parsed and tables that report no `numBytes` failed to split (Python) ([#39597](https://github.com/apache/beam/issues/39597)). +* (Python) Added `AwsOptions` with `--aws_*` pipeline options (e.g. `--aws_access_key_id`, `--aws_region_name`, `--aws_endpoint_url`) for configuring AWS service clients such as S3 ([#40528](https://github.com/apache/beam/issues/40528)). ## Breaking Changes @@ -84,7 +85,7 @@ ## Deprecations * Support for Python 3.10 was removed following its end-of-life ([#40250](https://github.com/apache/beam/issues/40250)). - +* (Python) `S3Options` and its `--s3_*` pipeline options are deprecated in favor of `AwsOptions` and the equivalent `--aws_*` options. The `--s3_*` options still work and are used when the corresponding `--aws_*` option is not set ([#40528](https://github.com/apache/beam/issues/40528)). * X behavior is deprecated and will be removed in X versions ([#X](https://github.com/apache/beam/issues/X)). ## Bugfixes