[Python] Add AwsOptions and deprecate S3Options

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

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(

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

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):
 

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

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):

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)