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 3ec3db07aa1 [Python] Make `--aws_*` options defaults and `--s3_*`
options S3 overrides (#40532)
3ec3db07aa1 is described below
commit 3ec3db07aa1b65da98c09129476abe23f1962fa8
Author: Shunping Huang <[email protected]>
AuthorDate: Mon Oct 12 21:53:40 2026 -0400
[Python] Make `--aws_*` options defaults and `--s3_*` options S3 overrides
(#40532)
* Make --aws_* options defaults and --s3_* options S3 overrides
* Reformat
---
CHANGES.md | 3 +-
.../apache_beam/io/aws/clients/s3/boto3_client.py | 33 ++--
.../io/aws/clients/s3/boto3_client_test.py | 17 ++-
.../python/apache_beam/options/pipeline_options.py | 169 ++++++++++++---------
.../apache_beam/options/pipeline_options_test.py | 84 +++++++---
5 files changed, 193 insertions(+), 113 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 354f971acc6..fda4c9edc21 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -73,7 +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)).
+* (Python) Added `AwsOptions` with `--aws_*` pipeline options (e.g.
`--aws_access_key_id`, `--aws_region_name`, `--aws_endpoint_url`) that set
defaults for all AWS service clients. The existing `--s3_*` options in
`S3Options` override these defaults for the S3 client only
([#40528](https://github.com/apache/beam/issues/40528)).
## Breaking Changes
@@ -86,7 +86,6 @@
## 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 9182c1d7df8..57c7c83f3ca 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
@@ -41,33 +41,32 @@ class Client(object):
Wrapper for boto3 library
"""
def __init__(self, options):
+ """
+ Args:
+ options: PipelineOptions, or a dict of option names (aws_* and s3_*) to
+ values. --s3_* options override the --aws_* defaults for S3.
+ """
assert boto3 is not None, 'Missing boto3 requirement'
if isinstance(options, pipeline_options.PipelineOptions):
aws_options = options.view_as(pipeline_options.AwsOptions)
+ s3_options = options.view_as(pipeline_options.S3Options)
else:
- # A plain dict of option names (aws_* or the deprecated s3_*) to values.
aws_options = pipeline_options.AwsOptions(flags=[], **options)
+ s3_options = pipeline_options.S3Options(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
+ settings = aws_options.get_client_settings(s3_options)
session = boto3.session.Session()
self.client = session.client(
service_name='s3',
- region_name=region_name,
- api_version=api_version,
- use_ssl=use_ssl,
- verify=verify,
- endpoint_url=endpoint_url,
- aws_access_key_id=access_key_id,
- aws_secret_access_key=secret_access_key,
- aws_session_token=session_token)
+ region_name=settings['region_name'],
+ api_version=s3_options.s3_api_version,
+ use_ssl=not settings['disable_ssl'],
+ verify=settings['verify'],
+ endpoint_url=settings['endpoint_url'],
+ aws_access_key_id=settings['access_key_id'],
+ aws_secret_access_key=settings['secret_access_key'],
+ aws_session_token=settings['session_token'])
self._download_request = None
self._download_stream = None
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
index ccea8a3c664..490d7747294 100644
--- 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
@@ -48,7 +48,7 @@ class Boto3ClientOptionsTest(unittest.TestCase):
'--aws_session_token=TOKEN',
'--aws_endpoint_url=http://localhost:4566',
'--aws_region_name=us-east-1',
- '--aws_api_version=2006-03-01',
+ '--s3_api_version=2006-03-01',
'--aws_verify=False',
'--aws_disable_ssl',
]))
@@ -66,7 +66,7 @@ class Boto3ClientOptionsTest(unittest.TestCase):
'aws_session_token': 'TOKEN',
})
- def test_deprecated_s3_options_still_work(self):
+ def test_s3_options_without_aws_options(self):
boto3_client.Client(
PipelineOptions([
'--s3_access_key_id=AKIDEXAMPLE',
@@ -80,13 +80,20 @@ class Boto3ClientOptionsTest(unittest.TestCase):
self.assertEqual(kwargs['region_name'], 'us-west-2')
self.assertFalse(kwargs['use_ssl'])
- def test_aws_options_take_precedence_over_s3_options(self):
+ def test_s3_options_override_aws_options(self):
boto3_client.Client(
PipelineOptions([
- '--s3_region_name=us-west-2',
+ '--aws_access_key_id=AKIDEXAMPLE',
'--aws_region_name=us-east-1',
+ '--aws_endpoint_url=https://vpce.example.com',
+ '--s3_region_name=us-west-2',
+ '--s3_endpoint_url=http://localhost:9000',
]))
- self.assertEqual(self._client_kwargs()['region_name'], 'us-east-1')
+ kwargs = self._client_kwargs()
+ self.assertEqual(kwargs['region_name'], 'us-west-2')
+ self.assertEqual(kwargs['endpoint_url'], 'http://localhost:9000')
+ # Settings without an S3 override use the --aws_* default.
+ self.assertEqual(kwargs['aws_access_key_id'], 'AKIDEXAMPLE')
def test_dict_options(self):
boto3_client.Client({
diff --git a/sdks/python/apache_beam/options/pipeline_options.py
b/sdks/python/apache_beam/options/pipeline_options.py
index 37a98b63820..24abcae4d3e 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -2399,30 +2399,27 @@ class OptionsContext(object):
return options
-# 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()
-}
+# Client settings shared by all AWS clients. Each is configured in AwsOptions
as
+# --aws_<setting>, and can be overridden for one service by a service options
+# class (see AwsOptions.get_client_settings), e.g. --s3_<setting> in S3Options.
+_AWS_CLIENT_SETTINGS = (
+ 'access_key_id',
+ 'secret_access_key',
+ 'session_token',
+ 'endpoint_url',
+ 'region_name',
+ 'verify',
+ 'disable_ssl',
+)
class AwsOptions(PipelineOptions):
- """Options used to create AWS service clients, e.g. for S3.
+ """Default options for creating AWS service clients.
- The deprecated ``--s3_*`` flags from :class:`S3Options` are still accepted.
- Each ``aws_*`` option falls back to its ``s3_*`` counterpart when it is not
- set.
+ These settings apply to every AWS client Beam creates. A service options
+ class can override any of them for its own client, e.g. ``--s3_endpoint_url``
+ in :class:`S3Options` overrides ``--aws_endpoint_url`` for S3 only. Use
+ :meth:`get_client_settings` to get the effective settings for a service.
"""
@classmethod
def _add_argparse_args(cls, parser):
@@ -2441,69 +2438,105 @@ class AwsOptions(PipelineOptions):
parser.add_argument(
'--aws_endpoint_url',
default=None,
- help='The complete URL to use for the constructed AWS client.')
+ help='The complete URL to use for the constructed AWS clients.')
parser.add_argument(
'--aws_region_name',
default=None,
- help='The name of the region associated with the AWS client.')
- parser.add_argument(
- '--aws_api_version',
- default=None,
- help='The API version to use with the AWS client.')
+ help='The name of the region associated with the AWS clients.')
parser.add_argument(
'--aws_verify',
default=None,
- help='Whether or not to verify SSL certificates with the AWS client.')
+ help='Whether or not to verify SSL certificates with the AWS clients.')
parser.add_argument(
'--aws_disable_ssl',
default=False,
action='store_true',
help=(
- 'Whether or not to use SSL with the AWS client. '
+ 'Whether or not to use SSL with the AWS clients. '
'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
+ def get_client_settings(
+ self,
+ service_options: Optional['PipelineOptions'] = None) -> dict[str, Any]:
+ """Returns the effective client settings for an AWS service.
+ Starts from the ``--aws_*`` defaults and applies the overrides set in
+ ``service_options``. A service options class opts in by defining
+ ``AWS_CLIENT_SETTINGS_PREFIX`` (e.g. ``'s3'``) and options named
+ ``<prefix>_<setting>`` for the settings it can override (e.g.
+ ``s3_endpoint_url``). An override applies when it is set (not None/False).
-class S3Options(AwsOptions):
- """Deprecated, use :class:`AwsOptions` instead.
+ Args:
+ service_options: Optional options view for a specific service, e.g.
+ ``options.view_as(S3Options)``.
- The ``s3_*`` attributes delegate to the corresponding ``aws_*`` options, so
- both the old and new flags are honored.
+ Returns:
+ A dict keyed by setting name (``access_key_id``, ``secret_access_key``,
+ ``session_token``, ``endpoint_url``, ``region_name``, ``verify``,
+ ``disable_ssl``).
+ """
+ settings = {
+ name: getattr(self, 'aws_' + name)
+ for name in _AWS_CLIENT_SETTINGS
+ }
+ prefix = getattr(service_options, 'AWS_CLIENT_SETTINGS_PREFIX', None)
+ if prefix:
+ for name in _AWS_CLIENT_SETTINGS:
+ override = getattr(service_options, '%s_%s' % (prefix, name), None)
+ if override:
+ settings[name] = override
+ return settings
+
+
+class S3Options(PipelineOptions):
+ """Options for the S3 client.
+
+ ``--s3_api_version`` is S3-specific. The other options override the
+ corresponding ``--aws_*`` default in :class:`AwsOptions` for S3 only, e.g. to
+ point S3 at an S3-compatible endpoint while other AWS clients use the
+ defaults.
"""
- 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)
+ AWS_CLIENT_SETTINGS_PREFIX = 's3'
- 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)
+ @classmethod
+ def _add_argparse_args(cls, parser):
+ parser.add_argument(
+ '--s3_access_key_id',
+ default=None,
+ help='The access key ID for the S3 client. Overrides '
+ '--aws_access_key_id.')
+ parser.add_argument(
+ '--s3_secret_access_key',
+ default=None,
+ help='The secret access key for the S3 client. Overrides '
+ '--aws_secret_access_key.')
+ parser.add_argument(
+ '--s3_session_token',
+ default=None,
+ help='The session token for the S3 client. Overrides '
+ '--aws_session_token.')
+ parser.add_argument(
+ '--s3_endpoint_url',
+ default=None,
+ help='The complete URL to use for the S3 client. Overrides '
+ '--aws_endpoint_url.')
+ parser.add_argument(
+ '--s3_region_name',
+ default=None,
+ help='The name of the region associated with the S3 client. '
+ 'Overrides --aws_region_name.')
+ parser.add_argument(
+ '--s3_verify',
+ default=None,
+ help='Whether or not to verify SSL certificates with the S3 client. '
+ 'Overrides --aws_verify.')
+ parser.add_argument(
+ '--s3_disable_ssl',
+ default=False,
+ action='store_true',
+ help='Disable SSL for the S3 client, even if --aws_disable_ssl is not '
+ 'set.')
+ parser.add_argument(
+ '--s3_api_version',
+ default=None,
+ help='The API version to use with the S3 client.')
diff --git a/sdks/python/apache_beam/options/pipeline_options_test.py
b/sdks/python/apache_beam/options/pipeline_options_test.py
index f02ec10a69a..493e42d5301 100644
--- a/sdks/python/apache_beam/options/pipeline_options_test.py
+++ b/sdks/python/apache_beam/options/pipeline_options_test.py
@@ -1135,44 +1135,86 @@ class AwsOptionsTest(unittest.TestCase):
self.assertTrue(options.aws_disable_ssl)
self.assertIsNone(options.aws_endpoint_url)
- def test_aws_options_fall_back_to_deprecated_s3_flags(self):
+ def test_s3_options(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)
+ '--s3_api_version=2006-03-01',
+ ]).view_as(S3Options)
+ self.assertEqual(options.s3_access_key_id, 'key')
+ self.assertEqual(options.s3_region_name, 'us-west-2')
+ self.assertEqual(options.s3_api_version, '2006-03-01')
+ self.assertFalse(options.s3_disable_ssl)
- def test_aws_flags_take_precedence_over_s3_flags(self):
+ def test_client_settings_without_service_options(self):
options = PipelineOptions([
- '--s3_region_name=us-west-2',
+ '--aws_access_key_id=key',
'--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')
+ options.view_as(AwsOptions).get_client_settings(),
+ {
+ 'access_key_id': 'key',
+ 'secret_access_key': None,
+ 'session_token': None,
+ 'endpoint_url': None,
+ 'region_name': 'us-east-1',
+ 'verify': None,
+ 'disable_ssl': False,
+ })
- def test_s3_options_delegate_to_aws_options(self):
+ def test_s3_options_override_aws_defaults(self):
options = PipelineOptions([
'--aws_access_key_id=key',
+ '--aws_region_name=us-east-1',
+ '--aws_endpoint_url=https://vpce.example.com',
'--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)
+ '--s3_endpoint_url=http://localhost:9000',
+ '--s3_disable_ssl',
+ ])
+ settings = options.view_as(AwsOptions).get_client_settings(
+ options.view_as(S3Options))
+ self.assertEqual(settings['region_name'], 'us-west-2')
+ self.assertEqual(settings['endpoint_url'], 'http://localhost:9000')
+ self.assertTrue(settings['disable_ssl'])
+ # Not overridden for S3, so the --aws_* default applies.
+ self.assertEqual(settings['access_key_id'], 'key')
+
+ def test_s3_options_from_kwargs(self):
+ options = PipelineOptions(s3_endpoint_url='http://localhost:4566')
+ settings = options.view_as(AwsOptions).get_client_settings(
+ options.view_as(S3Options))
+ self.assertEqual(settings['endpoint_url'], 'http://localhost:4566')
+
+ def test_other_services_can_override_aws_defaults(self):
+ # Any service options class can override the defaults by declaring a
+ # prefix and <prefix>_<setting> options, e.g. a future
+ # SecretsManagerOptions with --secretsmanager_endpoint_url.
+ class FakeServiceOptions(object):
+ AWS_CLIENT_SETTINGS_PREFIX = 'fake'
+ fake_endpoint_url = 'https://fake.example.com'
+ fake_region_name = None
+
+ aws_options = PipelineOptions([
+ '--aws_region_name=us-east-1',
+ '--aws_endpoint_url=https://default.example.com',
+ ]).view_as(AwsOptions)
+ settings = aws_options.get_client_settings(FakeServiceOptions())
+ self.assertEqual(settings['endpoint_url'], 'https://fake.example.com')
+ self.assertEqual(settings['region_name'], 'us-east-1')
+
+ def test_s3_api_version_is_s3_only(self):
+ options = PipelineOptions(['--s3_api_version=2006-03-01'])
+ self.assertNotIn('aws_api_version', options.get_all_options())
+ self.assertNotIn(
+ 'api_version', options.view_as(AwsOptions).get_client_settings())
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)
+ self.assertIn('s3_api_version', all_options)
if __name__ == '__main__':
