This is an automated email from the ASF dual-hosted git repository.
shunping 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 d85fbc50cf9 [Python] Add `AwsOptions` and deprecate `S3Options`
(#40529)
d85fbc50cf9 is described below
commit d85fbc50cf91fa5b181b33664563dd32cd0364db
Author: Shunping Huang <[email protected]>
AuthorDate: Mon Oct 12 12:32:10 2026 -0400
[Python] Add `AwsOptions` and deprecate `S3Options` (#40529)
---
CHANGES.md | 3 +-
.../apache_beam/io/aws/clients/s3/boto3_client.py | 29 +++---
.../io/aws/clients/s3/boto3_client_test.py | 111 +++++++++++++++++++++
.../apache_beam/io/aws/clients/s3/client_test.py | 2 +-
sdks/python/apache_beam/io/aws/s3filesystem.py | 2 +-
sdks/python/apache_beam/io/aws/s3io_test.py | 2 +-
.../python/apache_beam/options/pipeline_options.py | 106 ++++++++++++++++----
.../apache_beam/options/pipeline_options_test.py | 54 ++++++++++
8 files changed, 270 insertions(+), 39 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index cbda51c7c60..0525ff92792 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
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 aee24ac2c05..9182c1d7df8 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 00000000000..ccea8a3c664
--- /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
+
+
[email protected](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 67797b1736b..ea110ea9b5d 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 @@ class ClientErrorTest(unittest.TestCase):
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 584263ec241..f7ec8a28d2b 100644
--- a/sdks/python/apache_beam/io/aws/s3filesystem.py
+++ b/sdks/python/apache_beam/io/aws/s3filesystem.py
@@ -41,7 +41,7 @@ class S3FileSystem(FileSystem):
"""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 09df3da7674..ee871be9d82 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 @@ class TestS3IO(unittest.TestCase):
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 743ab967b04..37a98b63820 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -51,6 +51,7 @@ __all__ = [
'ProfilingOptions',
'SetupOptions',
'TestOptions',
+ 'AwsOptions',
'S3Options'
]
@@ -2398,42 +2399,111 @@ class OptionsContext(object):
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 d9e1d850ec7..f02ec10a69a 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 hamcrest as hc
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 @@ class CompatVersionTest(unittest.TestCase):
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()
