Airflow Summit 2023
Unisciti alla community di Airflow dal 19 al 21 settembre durante la conferenza Airflow Summit 2023 per scoprire di più su Airflow e condividere la tua esperienza. La chiamata per i documenti è aperta

Accedere all'API REST Airflow

Cloud Composer 1 | Cloud Composer 2

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

Per un esempio di utilizzo dell'API REST Airflow con Cloud Functions, consulta Attivazione dei DAG con Cloud Functions.

Versioni API Airflow REST

In Cloud Composer 2 sono disponibili le seguenti versioni dell'API REST Airflow:

Airflow 2 utilizza l'API REST stabile. L'API REST sperimentale è deprecata da Airflow.
Puoi comunque utilizzare l'API REST sperimentale in Airflow 2 se esegui l'abilitazione tramite un override della configurazione di Airflow, come descritto più avanti.

Prima di iniziare

Attiva l'API Cloud Composer.

Abilita l'API

Abilita l'API REST Airflow stabile

L'API REST stabile è già abilitata per impostazione predefinita in Airflow 2.

Cloud Composer utilizza il proprio backend per l'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 stabile o modificare il ruolo utente predefinito ignorando le seguenti opzioni di configurazione di Airflow:

Sezione	Chiave	Valore	Note
`api`	(Airflow 2.2.5 e versioni precedenti) `auth_backend` (Airflow 2.3.0 e versioni successive) `auth_backends`	`airflow.composer.api.backend.composer_auth`	Per disabilitare l'API REST stabile, passa a `airflow.api.auth.backend.deny_all`
`api`	`composer_auth_user_registration_role`	`Op`	Puoi specificare qualsiasi altro ruolo.

Attiva l'API REST Airflow sperimentale

Per impostazione predefinita, la funzionalità di autenticazione API è disattivata nell'API sperimentale. Il server web di Airflow nega tutte le richieste che fai.

Per abilitare la funzionalità di autenticazione API e l'API sperimentale Airflow 2, sostituisci la seguente opzione di configurazione Airflow:

Sezione	Chiave	Valore	Note
`api`	(Airflow 2.2.5 e versioni precedenti) `auth_backend` (Airflow 2.3.0 e versioni successive) `auth_backends`	`airflow.api.auth.backend.default`	Il valore predefinito è `airflow.composer.api.backend.composer_auth`.
`api`	`enable_experimental_api`	`True`	Il valore predefinito è `False`.

Dopo aver impostato questa opzione di configurazione su airflow.api.auth.backend.default, il server web di Airflow accetta tutte le richieste API senza autenticazione. Anche se il server web di Airflow non richiede l'autenticazione, è comunque protetto da Identity-Aware Proxy, che fornisce il proprio livello di autenticazione.

Consenti 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 del chiamante può utilizzare l'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, che è 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 stabile.

Inserisci il contenuto dell'esempio seguente in un file denominato composer2_airflow_rest_api.py, quindi fornisci l'URL dell'interfaccia utente di Airflow, il nome del DAG e la configurazione di esecuzione del DAG nei valori della variabile.

composer/rest/composer2/composer2_airflow_rest_api.py

Visualizza su GitHub Feedback

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)

Ad esempio, la configurazione riportata di seguito non è corretta

  web_server_url = (
    "https://example-airflow-ui-url-dot-us-central1.composer.googleusercontent.com/"
  )

Accedere all'API REST Airflow utilizzando un account di servizio

Il database Airflow limita la lunghezza del campo dell'email a 64 caratteri. A volte gli account di servizio hanno indirizzi email che superano i 64 caratteri. Non è possibile creare utenti Airflow per questi account di servizio nel modo consueto. Se non esiste un utente Airflow per questo account di servizio, l'accesso all'API Airflow REST causa errori HTTP 401 e 403.

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

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 dell'account di servizio.
- PROJECT_ID con l'ID progetto.
Crea un utente Airflow con il ruolo Op per l'account di servizio:
UI di Airflow
1. Vai all'interfaccia utente di Airflow. L'utente Airflow deve avere il ruolo Admin.
2. Vai a Sicurezza > Elenca utenti e fai clic su Aggiungi un nuovo record.
3. Specifica accounts.google.com:NUMERIC_USER_ID come nome utente. Sostituisci NUMERIC_USER_ID con lo User-ID ottenuto nel passaggio precedente.
4. Specifica un identificatore univoco come indirizzo email. Puoi utilizzare qualsiasi stringa univoca.
5. Specifica il ruolo dell'utente. Ad esempio, Op.
6. Assicurati che la casella di controllo È attiva? 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 lo User-ID ottenuto nel passaggio precedente.
- UNIQUE_ID con l'identificatore per l'utente Airflow. Puoi utilizzare qualsiasi stringa univoca.
Dopo aver creato un utente Airflow per un account di servizio, un chiamante autenticato come account di servizio viene riconosciuto come utente preregistrato e ha eseguito l'accesso a Airflow.

Scalabilità dell'API REST Airflow in scala

L'API REST Airflow e gli endpoint UI di Airflow vengono eseguiti all'interno del componente, ovvero il server web di Airflow. Se utilizzi l'API REST in modo intensivo, valuta la possibilità di aumentare i parametri di CPU e memoria per regolare le risorse Webserver Airflow in base al carico previsto.