diff --git a/CHANGES.md b/CHANGES.md index 0525ff927920..f975856649da 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 @@ -85,7 +85,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 9182c1d7df80..57c7c83f3caf 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 ccea8a3c6640..490d7747294f 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 @@ def test_aws_options_passed_to_client(self): '--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 @@ def test_aws_options_passed_to_client(self): '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 @@ def test_deprecated_s3_options_still_work(self): 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 37a98b638209..24abcae4d3e3 100644 --- a/sdks/python/apache_beam/options/pipeline_options.py +++ b/sdks/python/apache_beam/options/pipeline_options.py @@ -2399,30 +2399,27 @@ def augment_options(cls, options): 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_, and can be overridden for one service by a service options +# class (see AwsOptions.get_client_settings), e.g. --s3_ 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 @@ def _add_argparse_args(cls, parser): 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 + ``_`` 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 f02ec10a69a4..493e42d53010 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 @@ def test_aws_options(self): 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 _ 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__':