Utilizzo di una chiave segreta Gestione dei segreti AWS 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 Gestione dei segreti AWS per una variabile Apache Airflow

I seguenti esempi di chiamate Gestione dei segreti AWS 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 Gestione dei segreti AWS

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