Accedere all'API REST Airflow

Managed Airflow (terza generazione) | Managed Airflow (seconda generazione) | Managed Airflow (prima generazione legacy)

Apache Airflow ha un'interfaccia API REST che puoi utilizzare per eseguire attività come ottenere informazioni sulle esecuzioni e sulle attività dei DAG, aggiornare i DAG, ottenere la configurazione di Airflow, aggiungere ed eliminare connessioni ed elencare gli utenti.

Per un esempio di utilizzo dell'API REST Airflow con Cloud Run Functions, consulta Attivare i DAG con Cloud Run Functions.

Versioni dell'API REST Airflow

• Airflow 3 utilizza l'API REST Airflow v2.
• Airflow 2 utilizza l'API REST Airflow v1.

Configurare l'API REST Airflow

Airflow 3

Airflow 3 utilizza l'API REST Airflow v2.

Managed Airflow utilizza il proprio backend di autenticazione API.

L'autorizzazione funziona nel modo standard fornito da Airflow 3. Quando un nuovo utente autorizza tramite l'API, il suo account riceve il ruolo Op per impostazione predefinita.

L'API REST Airflow in Airflow 3 è sempre abilitata e non è possibile disabilitarla. Puoi modificare il ruolo utente predefinito sostituendo la seguente opzione di configurazione di Airflow:

Sezione	Chiave	Valore	Note
api	composer_auth_user_registration_role	Op	Puoi specificare qualsiasi altro ruolo.

Airflow 2

L'API REST Airflow v1 è abilitata per impostazione predefinita in Airflow 2.

Managed Airflow utilizza il proprio backend di autenticazione API.

L'autorizzazione funziona nel modo standard fornito da Airflow. Quando un nuovo utente autorizza tramite l'API, il suo account riceve il ruolo Op per impostazione predefinita.

Puoi abilitare o disabilitare l'API REST Airflow o modificare il ruolo utente predefinito sostituendo le seguenti opzioni di configurazione di Airflow:

Sezione	Chiave	Valore	Note
api	auth_backends	airflow.composer.api.backend.composer_auth	Per disabilitare l'API REST Airflow, imposta airflow.api.auth.backend.deny_all
api	composer_auth_user_registration_role	Op	Puoi specificare qualsiasi altro ruolo.

Consentire le chiamate API all'API REST Airflow utilizzando il controllo dell'accesso al server web

A seconda del metodo utilizzato per chiamare l'API REST Airflow, il metodo chiamante può utilizzare un indirizzo IPv4 o IPv6. Ricorda di sbloccare il traffico IP verso l'API REST Airflow utilizzando il controllo dell'accesso al server web.

Utilizza l'opzione di configurazione predefinita All IP addresses have access (default) se non sai da quali indirizzi IP verranno inviate le chiamate all'API REST Airflow.

Effettuare chiamate all'API REST Airflow

Questa sezione fornisce uno script Python di esempio che puoi utilizzare per attivare i DAG con l'API REST Airflow.

Nello script, imposta le seguenti variabili:

• dag_id: nome di un DAG, come definito nel file di origine del DAG.
• dag_config: configurazione per l'esecuzione del DAG.

• web_server_url: l'URL del server web Airflow. Il formato è https://<web-server-id>.composer.googleusercontent.com.

  Importante: non inserire la barra rovesciata alla fine dell'URL del server web.

• (Airflow 3) logical_date: la data logica per l'esecuzione del DAG.

Airflow 3

"""
Trigger a DAG in Cloud Composer 3 environment with Airflow 3 using the Airflow REST API v2.
"""

from __future__ import annotations

from typing import Any

import google.auth
from google.auth.transport.requests import AuthorizedSession
import requests

# Following best practices, these credentials should be
# constructed at start-up time and used throughout
# https://cloud.google.com/apis/docs/client-libraries-best-practices
AUTH_SCOPE = "https://www.googleapis.com/auth/cloud-platform"
CREDENTIALS, _ = google.auth.default(scopes=[AUTH_SCOPE])

def make_composer3_web_server_request(
    url: str, method: str = "GET", **kwargs: Any
) -> google.auth.transport.Response:
    """
    Make a request to Cloud Composer 3 environment's web server with Airflow 3.
    Args:
      url: The URL to fetch.
      method: The request method to use ('GET', 'OPTIONS', 'HEAD', 'POST', 'PUT',
        'PATCH', 'DELETE')
      **kwargs: Any of the parameters defined for the request function:
                https://github.com/requests/requests/blob/master/requests/api.py
                  If no timeout is provided, it is set to 90 by default.
    """

    authed_session = AuthorizedSession(CREDENTIALS)

    # Set the default timeout, if missing
    if "timeout" not in kwargs:
        kwargs["timeout"] = 90

    return authed_session.request(method, url, **kwargs)

def trigger_dag(web_server_url: str, dag_id: str, data: dict, logical_date: str) -> str:
    """
    Make a request to trigger a DAG using the Airflow REST API v2.
    https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html

    Args:
      web_server_url: The URL of the Airflow 3 web server.
      dag_id: The DAG ID.
      data: Additional configuration parameters for the DAG run (json).
    """

    endpoint = f"api/v2/dags/{dag_id}/dagRuns"
    request_url = f"{web_server_url}/{endpoint}"
    json_data = {"conf": data, "logical_date": logical_date}

    response = make_composer3_web_server_request(
        request_url, method="POST", json=json_data
    )

    if response.status_code == 403:
        raise requests.HTTPError(
            "You do not have a permission to perform this operation. "
            "Check Airflow RBAC roles for your account."
            f"{response.headers} / {response.text}"
        )
    elif response.status_code != 200:
        response.raise_for_status()
    else:
        return response.text

if __name__ == "__main__":
    # TODO(developer): replace with your values
    dag_id = "airflow_monitoring"  # Replace with the ID of the DAG that you want to run.
    dag_config = {
        "your-key": "your-value"
    }  # Replace with configuration parameters for the DAG run.
    # Replace web_server_url with the Airflow web server address. To obtain this
    # URL, run the following command for your environment:
    # gcloud composer environments describe example-environment \
    #  --location=your-composer-region \
    #  --format="value(config.airflowUri)"
    logical_date = "2025-01-01T14:00:00Z" # Replace with the data interval for which to run the DAG
    web_server_url = (
        "https://example-airflow-ui-url-dot-us-central1.composer-staging.googleusercontent.com"
    )

    response_text = trigger_dag(
        web_server_url=web_server_url, dag_id=dag_id, data=dag_config, logical_date=logical_date
    )

    print(response_text)

Airflow 2

from __future__ import annotations

from typing import Any

import google.auth
from google.auth.transport.requests import AuthorizedSession
import requests


# Following GCP best practices, these credentials should be
# constructed at start-up time and used throughout
# https://cloud.google.com/apis/docs/client-libraries-best-practices
AUTH_SCOPE = "https://www.googleapis.com/auth/cloud-platform"
CREDENTIALS, _ = google.auth.default(scopes=[AUTH_SCOPE])


def make_composer2_web_server_request(
    url: str, method: str = "GET", **kwargs: Any
) -> google.auth.transport.Response:
    """
    Make a request to Cloud Composer 2 environment's web server.
    Args:
      url: The URL to fetch.
      method: The request method to use ('GET', 'OPTIONS', 'HEAD', 'POST', 'PUT',
        'PATCH', 'DELETE')
      **kwargs: Any of the parameters defined for the request function:
                https://github.com/requests/requests/blob/master/requests/api.py
                  If no timeout is provided, it is set to 90 by default.
    """

    authed_session = AuthorizedSession(CREDENTIALS)

    # Set the default timeout, if missing
    if "timeout" not in kwargs:
        kwargs["timeout"] = 90

    return authed_session.request(method, url, **kwargs)


def trigger_dag(web_server_url: str, dag_id: str, data: dict) -> str:
    """
    Make a request to trigger a dag using the stable Airflow 2 REST API.
    https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html

    Args:
      web_server_url: The URL of the Airflow 2 web server.
      dag_id: The DAG ID.
      data: Additional configuration parameters for the DAG run (json).
    """

    endpoint = f"api/v1/dags/{dag_id}/dagRuns"
    request_url = f"{web_server_url}/{endpoint}"
    json_data = {"conf": data}

    response = make_composer2_web_server_request(
        request_url, method="POST", json=json_data
    )

    if response.status_code == 403:
        raise requests.HTTPError(
            "You do not have a permission to perform this operation. "
            "Check Airflow RBAC roles for your account."
            f"{response.headers} / {response.text}"
        )
    elif response.status_code != 200:
        response.raise_for_status()
    else:
        return response.text




if __name__ == "__main__":
    # TODO(developer): replace with your values
    dag_id = "your-dag-id"  # Replace with the ID of the DAG that you want to run.
    dag_config = {
        "your-key": "your-value"
    }  # Replace with configuration parameters for the DAG run.
    # Replace web_server_url with the Airflow web server address. To obtain this
    # URL, run the following command for your environment:
    # gcloud composer environments describe example-environment \
    #  --location=your-composer-region \
    #  --format="value(config.airflowUri)"
    web_server_url = (
        "https://example-airflow-ui-url-dot-us-central1.composer.googleusercontent.com"
    )

    response_text = trigger_dag(
        web_server_url=web_server_url, dag_id=dag_id, data=dag_config
    )

    print(response_text)

Accedere all'API REST Airflow utilizzando un account di servizio

Il database Airflow nelle versioni di Airflow precedenti alla 2.3.0 limita la lunghezza del campo email a 64 caratteri. A volte i service account hanno indirizzi email più lunghi di 64 caratteri. Non è possibile creare utenti Airflow per questi service account nel modo consueto. Se non esiste un utente Airflow per un account di servizio di questo tipo, l'accesso all'API REST Airflow genera errori HTTP 401 e 403.

Nota: gli errori 401 (Autorizzazione negata) e 403 (Accesso negato) sono errori HTTP generici. Se continui a ricevere questi errori dopo aver utilizzato la soluzione alternativa descritta, controlla altre possibili cause. Ad esempio, potrebbero mancare i ruoli IAM nel account di servizio. Per consentire a un account di servizio di accedere all'API REST Airflow, concedi un ruolo IAM che consenta l'accesso all'interfaccia web.

In alternativa, puoi preregistrare un utente Airflow per un account di servizio. Per farlo, utilizza accounts.google.com:NUMERIC_USER_ID come nome utente e una stringa univoca come email.

1. Per ottenere NUMERIC_USER_ID per un account di servizio, esegui:

   gcloud iam service-accounts describe \
     SA_NAME@PROJECT_ID.iam.gserviceaccount.com \
     --format="value(oauth2ClientId)"
   

   Sostituisci:

   • SA_NAME con il nome del account di servizio.
   • PROJECT_ID con l'ID progetto.

2. Crea un utente Airflow con il ruolo Op per il account di servizio:

   UI di Airflow

   1. Vai all'UI di Airflow.

   2. Vai a Sicurezza > Elenca utenti e fai clic Aggiungi un nuovo record. Per aprire questa pagina, l'utente Airflow deve avere il ruolo Admin.

   3. Specifica accounts.google.com:NUMERIC_USER_ID come nome utente. Sostituisci NUMERIC_USER_ID con l'ID utente ottenuto nel passaggio precedente.

   4. Specifica un identificatore univoco come email. Puoi utilizzare qualsiasi stringa univoca.

   5. Specifica il ruolo per l'utente. Ad esempio, Op.

   6. Assicurati che la casella di controllo È attivo? sia selezionata.

   7. Specifica il nome e il cognome dell'utente. Puoi utilizzare qualsiasi stringa.

   8. Fai clic su Salva.

   gcloud

   Esegui il seguente comando dell'interfaccia a riga di comando di Airflow:

   gcloud composer environments run ENVIRONMENT_NAME \
       --location LOCATION \
       users create -- \
       -u accounts.google.com:NUMERIC_USER_ID \
       -e UNIQUE_ID  \
       -f UNIQUE_ID \
       -l - -r Op --use-random-password
   

   Sostituisci:

   • ENVIRONMENT_NAME con il nome dell'ambiente.
   • LOCATION con la regione in cui si trova l'ambiente.
   • NUMERIC_USER_ID con l'ID utente ottenuto nel passaggio precedente.
   • UNIQUE_ID con l'identificatore per l'utente Airflow. Puoi utilizzare qualsiasi stringa univoca.

3. Dopo aver creato un utente Airflow per un account di servizio, un chiamante autenticato come account di servizio viene riconosciuto come utente preregistrato e accede ad Airflow.

Scalare il componente API REST Airflow

Gli endpoint dell'API REST Airflow e dell'UI di Airflow vengono eseguiti all'interno del server web Airflow. Se utilizzi l'API REST in modo intensivo, valuta la possibilità di aumentare la quantità di CPU e memoria disponibile per il server web Airflow, in base al carico previsto.

Passaggi successivi

• Attivare i DAG con Cloud Run Functions
• Interfaccia a riga di comando di Airflow