You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by sh...@apache.org on 2026/10/16 19:50:41 UTC
(beam) branch master updated: [Python] Add examples for AWS credentials from a secret manager (#40592)
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 e41f0ce1d1d [Python] Add examples for AWS credentials from a secret manager (#40592)
e41f0ce1d1d is described below
commit e41f0ce1d1d5526190e7e3de1820f5a7b22e05d3
Author: Shunping Huang <sh...@google.com>
AuthorDate: Fri Oct 16 15:50:33 2026 -0400
[Python] Add examples for AWS credentials from a secret manager (#40592)
* Add examples for AWS credentials from a secret manager
Add examples under apache_beam/examples/secret_manager that set
--aws_credentials_secret_manager to read AWS credentials from Google Cloud
Secret Manager:
* aws_credentials_s3_read.py: reads an S3 file outside a pipeline.
* aws_credentials_s3_pipeline.py: reads an S3 file in a pipeline
(DirectRunner or Dataflow).
* aws_credentials_secret_read.py: reads a secret from AWS Secrets Manager
with Secret.parse_secret_option and Secret.from_json.
* Add a use case of arn for reading AWS secrets
* Reformat
* Added a requirements file for the example.
---
.../examples/secret_manager/__init__.py | 18 +++++
.../aws_credentials_requirements.txt | 21 ++++++
.../secret_manager/aws_credentials_s3_pipeline.py | 75 +++++++++++++++++++
.../secret_manager/aws_credentials_s3_read.py | 60 +++++++++++++++
.../secret_manager/aws_credentials_secret_read.py | 87 ++++++++++++++++++++++
5 files changed, 261 insertions(+)
diff --git a/sdks/python/apache_beam/examples/secret_manager/__init__.py b/sdks/python/apache_beam/examples/secret_manager/__init__.py
new file mode 100644
index 00000000000..92dc81c246f
--- /dev/null
+++ b/sdks/python/apache_beam/examples/secret_manager/__init__.py
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+"""Examples for SecretManager integration in Apache Beam."""
diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_requirements.txt b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_requirements.txt
new file mode 100644
index 00000000000..e0db2ef261d
--- /dev/null
+++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_requirements.txt
@@ -0,0 +1,21 @@
+#
+# 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.
+#
+
+# Worker dependencies for aws_credentials_s3_pipeline.py: boto3 for the S3
+# filesystem, and google-cloud-secret-manager to resolve the AWS credentials.
+boto3>=1.9,<2
+google-cloud-secret-manager>=2.0,<3
diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py
new file mode 100644
index 00000000000..53544465349
--- /dev/null
+++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py
@@ -0,0 +1,75 @@
+#
+# 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.
+#
+
+"""A pipeline that reads an S3 file and logs each line, using AWS credentials
+stored in Google Cloud Secret Manager.
+
+Usage (DirectRunner):
+ python -m apache_beam.examples.secret_manager.aws_credentials_s3_pipeline \
+ --input=s3://my-bucket/my-file.txt \
+ --aws_region_name=us-east-1 \
+ --aws_credentials_secret_manager=GoogleCloudSecretManager \
+ --aws_access_key_id='{"name": "my-aws-key-id", "project": "my-project"}' \
+ --aws_secret_access_key='{"name": "my-aws-secret-key", "project": "my-project"}'
+
+Usage (DataflowRunner): add the following flags to the command above.
+ --runner=DataflowRunner \
+ --project=my-project \
+ --region=us-central1 \
+ --temp_location=gs://my-gcs-bucket/temp \
+ --requirements_file=apache_beam/examples/secret_manager/aws_credentials_requirements.txt
+
+where aws_credentials_requirements.txt (in the same directory as this example)
+lists the packages the workers need: boto3 and google-cloud-secret-manager.
+
+Required permissions:
+ * The AWS IAM user that owns the access key needs ``s3:ListBucket`` on the
+ bucket (``arn:aws:s3:::my-bucket``, used to match the input path) and
+ ``s3:GetObject`` on the object (``arn:aws:s3:::my-bucket/my-file.txt``),
+ for example through the ``AmazonS3ReadOnlyAccess`` managed policy or a
+ narrower custom policy. If the object is encrypted with a customer managed
+ KMS key, it also needs ``kms:Decrypt`` on that key.
+ * The Google Cloud account that resolves the credentials needs the Secret
+ Manager Secret Accessor role (``roles/secretmanager.secretAccessor``) on
+ the two secrets that hold the access key ID and secret access key. The
+ credentials are resolved both when the pipeline is constructed and where
+ the file is read, so this is the account running the script and, on
+ Dataflow, also the worker service account.
+"""
+
+import argparse
+import logging
+
+import apache_beam as beam
+from apache_beam.options.pipeline_options import PipelineOptions
+
+
+def main(argv=None):
+ parser = argparse.ArgumentParser()
+ parser.add_argument('--input', required=True, help='S3 path to read.')
+ known_args, pipeline_args = parser.parse_known_args(argv)
+
+ with beam.Pipeline(options=PipelineOptions(pipeline_args)) as p:
+ _ = (
+ p
+ | 'Read' >> beam.io.ReadFromText(known_args.input)
+ | 'Log' >> beam.Map(logging.info))
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ main()
diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_read.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_read.py
new file mode 100644
index 00000000000..6ef1c1ba804
--- /dev/null
+++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_read.py
@@ -0,0 +1,60 @@
+#
+# 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.
+#
+
+"""Reads an S3 file outside of a pipeline, using AWS credentials stored in
+Google Cloud Secret Manager.
+
+Usage:
+ python -m apache_beam.examples.secret_manager.aws_credentials_s3_read \
+ --input=s3://my-bucket/my-file.txt \
+ --aws_region_name=us-east-1 \
+ --aws_credentials_secret_manager=GoogleCloudSecretManager \
+ --aws_access_key_id='{"name": "my-aws-key-id", "project": "my-project"}' \
+ --aws_secret_access_key='{"name": "my-aws-secret-key", "project": "my-project"}'
+
+Required permissions:
+ * The AWS IAM user that owns the access key needs ``s3:GetObject`` on the
+ object (e.g. ``arn:aws:s3:::my-bucket/my-file.txt``), for example through
+ the ``AmazonS3ReadOnlyAccess`` managed policy or a narrower custom policy.
+ If the object is encrypted with a customer managed KMS key, it also needs
+ ``kms:Decrypt`` on that key.
+ * The Google Cloud account running this script (e.g. your Application
+ Default Credentials) needs the Secret Manager Secret Accessor role
+ (``roles/secretmanager.secretAccessor``) on the two secrets that hold the
+ access key ID and secret access key.
+"""
+
+import argparse
+
+from apache_beam.io.filesystems import FileSystems
+from apache_beam.options.pipeline_options import PipelineOptions
+
+
+def main(argv=None):
+ parser = argparse.ArgumentParser()
+ parser.add_argument('--input', required=True, help='S3 path to read.')
+ known_args, pipeline_args = parser.parse_known_args(argv)
+
+ # The S3 filesystem resolves the --aws_* credentials through the secret
+ # manager given by --aws_credentials_secret_manager.
+ FileSystems.set_options(PipelineOptions(pipeline_args))
+ with FileSystems.open(known_args.input) as f:
+ print(f.read().decode('utf-8'))
+
+
+if __name__ == '__main__':
+ main()
diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py
new file mode 100644
index 00000000000..bf3dadedeb2
--- /dev/null
+++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py
@@ -0,0 +1,87 @@
+#
+# 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.
+#
+
+"""Reads a secret from AWS Secrets Manager outside of a pipeline, using AWS
+credentials stored in Google Cloud Secret Manager.
+
+Usage:
+ python -m apache_beam.examples.secret_manager.aws_credentials_secret_read \
+ --secret_id=my-aws-secret \
+ --secret_arn=arn:aws:secretsmanager:us-east-1:<account-id>:secret:my-aws-secret-AbCdEf \
+ --aws_region_name=us-east-1 \
+ --aws_credentials_secret_manager=GoogleCloudSecretManager \
+ --aws_access_key_id='{"name": "my-aws-key-id", "project": "my-project"}' \
+ --aws_secret_access_key='{"name": "my-aws-secret-key", "project": "my-project"}'
+
+Required permissions:
+ * The AWS IAM user that owns the access key needs
+ ``secretsmanager:GetSecretValue`` on the secret (e.g.
+ ``arn:aws:secretsmanager:us-east-1:<account-id>:secret:my-aws-secret-*``),
+ for example through the ``AWSSecretsManagerClientReadOnlyAccess`` managed
+ policy or a narrower custom policy. If the secret is encrypted with a
+ customer managed KMS key instead of the default ``aws/secretsmanager`` key,
+ it also needs ``kms:Decrypt`` on that key.
+ * The Google Cloud account running this script (e.g. your Application
+ Default Credentials) needs the Secret Manager Secret Accessor role
+ (``roles/secretmanager.secretAccessor``) on the two secrets that hold the
+ access key ID and secret access key.
+"""
+
+import argparse
+
+from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.utils.secret import Secret
+
+
+def main(argv=None):
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ '--secret_id',
+ required=True,
+ help='Name of the secret in AWS Secrets Manager.')
+ parser.add_argument(
+ '--secret_arn', help='Optional ARN of the secret in AWS Secrets Manager.')
+ known_args, pipeline_args = parser.parse_known_args(argv)
+ options = PipelineOptions(pipeline_args)
+
+ # The AWS Secrets Manager client uses the --aws_* options, with credentials
+ # resolved through --aws_credentials_secret_manager.
+
+ # Option 1: a 'type:<secret_type>;<param>:<value>' secret string.
+ secret = Secret.parse_secret_option(
+ f'type:awssecret;name:{known_args.secret_id}', options)
+ print(f'Retrieved secret of length {len(secret.get_str())}')
+
+ # Option 2: a JSON spec and a secret manager name.
+ secret = Secret.from_json(
+ f'{{"name": "{known_args.secret_id}"}}', 'AwsSecretsManager', options)
+ print(f'Retrieved secret of length {len(secret.get_str())}')
+
+ # Options 3 and 4: the same as options 1 and 2, but identifying the secret by
+ # its ARN.
+ if known_args.secret_arn:
+ secret = Secret.parse_secret_option(
+ f'type:awssecret;arn:{known_args.secret_arn}', options)
+ print(f'Retrieved secret of length {len(secret.get_str())}')
+
+ secret = Secret.from_json(
+ f'{{"arn": "{known_args.secret_arn}"}}', 'AwsSecretsManager', options)
+ print(f'Retrieved secret of length {len(secret.get_str())}')
+
+
+if __name__ == '__main__':
+ main()