Usando uma chave secreta AWS Secrets Manager para uma variável do Apache Airflow

O exemplo a seguir chama AWS Secrets Manager para obter uma chave secreta para uma variável do Apache Airflow no Amazon Managed Workflows for Apache Airflow. Pressupõe-se que você tenha concluído as etapas em Configurando uma conexão Apache Airflow usando um segredo AWS Secrets Manager.

Version (Versão)

• O código de amostra nesta página pode ser usado com o Apache Airflow v1 em Python 3.7.

• Você pode usar o exemplo de código nesta página com o Apache Airflow v2 em Python 3.10.

Pré-requisitos

Para usar o código de amostra nesta página, você precisará do seguinte:

• O back-end do Secrets Manager como uma opção de configuração do Apache Airflow, conforme mostrado em Configurando uma conexão Apache Airflow usando um segredo AWS Secrets Manager.

• Uma string variável do Apache Airflow no Secrets Manager, conforme mostrado em Configurando uma conexão Apache Airflow usando um segredo AWS Secrets Manager.

Permissões

• As permissões do Secrets Manager são mostradas em Configurando uma conexão Apache Airflow usando um segredo AWS Secrets Manager.

Requisitos

• Para usar esse exemplo de código com o Apache Airflow v1, nenhuma dependência adicional é necessária. O código usa a instalação básica do Apache Airflow v1 em seu ambiente.

• Para usar esse exemplo de código com o Apache Airflow v2, nenhuma dependência adicional é necessária. O código usa a instalação básica do Apache Airflow v2 em seu ambiente.

Exemplo de código

As etapas a seguir descrevem como criar o DAG código que chama o Secrets Manager para obter o segredo.

1. No prompt de comando, navegue até o diretório em que seu DAG código está armazenado. Por exemplo:

   cd dags

2. Copie o conteúdo da amostra de código a seguir e salve localmente como secrets-manager-var.py.

   from airflow import DAG
   from airflow.operators.python_operator import PythonOperator
   from airflow.models import Variable
   from airflow.utils.dates import days_ago
   from datetime import timedelta
   import os
   DAG_ID = os.path.basename(__file__).replace(".py", "")
   DEFAULT_ARGS = {
       'owner': 'airflow',
       'depends_on_past': False,
       'email': ['[e-mail]'],
       'email_on_failure': False,
       'email_on_retry': False,
   }
   def get_variable_fn(**kwargs):
       my_variable_name = Variable.get("test-variable", default_var="undefined")
       print("my_variable_name: ", my_variable_name)
       return my_variable_name
   with DAG(
       dag_id=DAG_ID,
       default_args=DEFAULT_ARGS,
       dagrun_timeout=timedelta(hours=2),
       start_date=days_ago(1),
       schedule_interval='@once',
       tags=['variable']
   ) as dag:
       get_variable = PythonOperator(
           task_id="get_variable",
           python_callable=get_variable_fn,
           provide_context=True
       )

Próximas etapas

• Saiba como fazer o upload do DAG código neste exemplo para a dags pasta em seu bucket do Amazon S3 em. Como adicionar ou atualizar DAGs