Utilizzo di una chiave segreta AWS Secrets Manager per una variabile Apache Airflow - Amazon Managed Workflows for Apache Airflow
VersionePrerequisitiAutorizzazioniRequisitiEsempio di codiceFasi successive

Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Utilizzo di una chiave segreta AWS Secrets Manager per una variabile Apache Airflow

I seguenti esempi di chiamate AWS Secrets Manager per ottenere una chiave segreta per una variabile Apache Airflow su Amazon Managed Workflows for Apache Airflow. Si presuppone che tu abbia completato i passaggi di. Configurazione di una connessione Apache Airflow utilizzando un segreto AWS Secrets Manager

Versione

È possibile utilizzare l'esempio di codice in questa pagina con Apache Airflow v2 in Python 3.10 e Apache Airflow v3in Python 3.11.

Prerequisiti

Per utilizzare il codice di esempio in questa pagina, avrai bisogno di quanto segue:

Autorizzazioni

Requisiti

Per utilizzare questo esempio di codice con Apache Airflow v2 e versioni successive, non sono necessarie dipendenze aggiuntive. Utilizzare aws-mwaa-docker-imagesper installare Apache Airflow.

Esempio di codice

I passaggi seguenti descrivono come creare il codice DAG che chiama Secrets Manager per ottenere il segreto.

  1. Nel prompt dei comandi, accedete alla directory in cui è memorizzato il codice DAG. Ad esempio:

    cd dags

  2. Copia il contenuto del seguente esempio di codice e salvalo localmente come. 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
    )

Fasi successive