使用秘密金鑰AWS Secrets Manager對於一個阿帕奇氣流連接

下列範例呼叫AWS Secrets Manager在亞馬遜 Apache 氣流管理的工作流程上獲取 Apache 氣流連接的秘密密鑰。它假設您已完成中的步驟使用 AWS Secrets Manager 密碼設定 Apache 氣流連線。

版本

• 此頁面上的範例程式碼可搭配使用阿帕奇氣流 V1在蟒蛇 3.7。

• 您可以使用此頁面上的程式碼範例阿帕奇氣流 v2 及以上在蟒蛇。

先決條件

若要使用此頁面上的範例程式碼，您需要下列項目：

• 秘密管理器後端作為 Apache 氣流配置選項，如圖所示使用 AWS Secrets Manager 密碼設定 Apache 氣流連線。

• 秘密管理器中的 Apache 氣流連接字符串，如圖所示使用 AWS Secrets Manager 密碼設定 Apache 氣流連線。

許可

• 密碼管理員權限，如所示使用 AWS Secrets Manager 密碼設定 Apache 氣流連線。

請求

• 若要將此程式碼範例與 Apache Airflow v1 搭配使用，不需要額外的相依性。該代碼使用阿帕奇氣流 V1 基本安裝在您的環境中。

• 若要將此程式碼範例與 Apache Airflow v2 搭配使用，不需要額外的相依性。該代碼使用阿帕奇氣流 V2 基地安裝在您的環境中。

程式碼範例

下列步驟說明如何建立呼叫秘密管理員以取得密碼的 DAG 程式碼。

Apache Airflow v2

1. 在命令提示字元中，瀏覽至儲存 DAG 程式碼的目錄。例如：

   cd dags

2. 複製下列程式碼範例的內容，並在本機儲存為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('secretsmanager')
       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
       )
   

Apache Airflow v1

1. 在命令提示字元中，瀏覽至儲存 DAG 程式碼的目錄。例如：

   cd dags

2. 複製下列程式碼範例的內容，並在本機儲存為secrets-manager.py。

   from airflow import DAG, settings, secrets
   from airflow.operators.python_operator import PythonOperator
   from airflow.utils.dates import days_ago
   from airflow.contrib.hooks.aws_hook import AwsHook
   
   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 = AwsHook()
       client = hook.get_client_type('secretsmanager')
       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 程式碼上傳至dags您的亞馬遜 S3 存儲桶中的文件夾新增或更新 DAG。