本文為英文版的機器翻譯版本,如內容有任何歧義或不一致之處,概以英文版為準。
在 中使用私密金鑰 AWS Secrets Manager 進行 Apache Airflow 連線
下列範例呼叫 AWS Secrets Manager 會在 Amazon Managed Workflows for Apache Airflow 上取得 Apache Airflow 連線的私密金鑰。它假設您已完成 中的步驟使用 AWS Secrets Manager 秘密設定 Apache Airflow 連線。
版本
您可以使用此頁面上的程式碼範例搭配 Python 3.10 中的 Apache Airflow v2 和 Python 3.11 中的 Apache Airflow v3。 https://peps.python.org/pep-0619/
先決條件
若要使用此頁面上的範例程式碼,您需要下列項目:
-
Secrets Manager 後端做為 Apache Airflow 組態選項,如 所列使用 AWS Secrets Manager 秘密設定 Apache Airflow 連線。
-
Secrets Manager 中的 Apache Airflow 連線字串,如 所列使用 AWS Secrets Manager 秘密設定 Apache Airflow 連線。
權限
-
Secrets Manager 許可,如 所列使用 AWS Secrets Manager 秘密設定 Apache Airflow 連線。
要求
若要搭配 Apache Airflow v2 和更新版本使用此程式碼範例,不需要額外的相依性。使用 aws-mwaa-docker-images
範例程式碼
下列步驟說明如何建立 DAG 程式碼,呼叫 Secrets Manager 來取得秘密。
-
在命令提示中,導覽至存放 DAG 程式碼的目錄。例如:
cd dags
-
複製下列程式碼範例的內容,並在本機儲存為
secrets-manager.py
。""" Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ from airflow import DAG, settings, secrets from airflow.operators.python import PythonOperator from airflow.utils.dates import days_ago from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook from datetime import timedelta import os ### The steps to create this secret key can be found at: https://docs.aws.amazon.com/mwaa/latest/userguide/connections-secrets-manager.html sm_secretId_name = 'airflow/connections/myconn' default_args = { 'owner': 'airflow', 'start_date': days_ago(1), 'depends_on_past': False } ### Gets the secret myconn from Secrets Manager def read_from_aws_sm_fn(**kwargs): ### set up Secrets Manager hook = AwsBaseHook(client_type='secretsmanager') client = hook.get_client_type(region_name='us-east-1') response = client.get_secret_value(SecretId=sm_secretId_name) myConnSecretString = response["SecretString"] return myConnSecretString ### 'os.path.basename(__file__).replace(".py", "")' uses the file name secrets-manager.py for a DAG ID of secrets-manager with DAG( dag_id=os.path.basename(__file__).replace(".py", ""), default_args=default_args, dagrun_timeout=timedelta(hours=2), start_date=days_ago(1), schedule_interval=None ) as dag: write_all_to_aws_sm = PythonOperator( task_id="read_from_aws_sm", python_callable=read_from_aws_sm_fn, provide_context=True )
後續步驟?
-
了解如何在此範例中將 DAG 程式碼上傳至 Amazon S3 儲存貯體中的
dags
資料夾新增或更新 DAGs。