Usando uma chave secreta AWS Secrets Manager para uma variável do Apache Airflow - Amazon Managed Workflows for Apache Airflow
VersãoPré-requisitosPermissõesRequisitosExemplo de códigoPróximas etapas

As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

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

O exemplo a seguir solicita AWS Secrets Manager a obtenção de 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 do Apache Airflow usando um segredo AWS Secrets Manager.

Versão

Você pode usar o exemplo de código nesta página com o Apache Airflow v2 no Python 3.10 e o Apache Airflow v3no Python 3.11.

Pré-requisitos

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

Permissões

Requisitos

Para usar esse exemplo de código com o Apache Airflow v2 e versões posteriores, não são necessárias dependências adicionais. Use aws-mwaa-docker-imagespara instalar o Apache Airflow.

Exemplo de código

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

  1. No prompt de comando, navegue até o diretório em que o código do DAG 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': ['airflow@example.com'],
    '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