[Python] Make --aws_* options defaults and --s3_* options S3 overrides

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

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

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({

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_<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 @@ 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
+    ``<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.')