Accedi all'API REST Airflow

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Apache Airflow ha un'interfaccia API REST che puoi utilizzare per eseguire attività come come ottenere informazioni sulle esecuzioni e attività dei DAG, aggiornare i DAG, ottenere Configurazione di Airflow, aggiunta ed eliminazione di connessioni ed elenco di utenti.

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

Versioni API REST Airflow

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

• Utilizzo di Airflow 2 l'API REST stabile. L'API REST sperimentale è deprecata da Airflow.

• Puoi comunque utilizzare l'API REST sperimentale in Airflow 2 se la abiliti attraverso un override della configurazione Airflow, come descritto ulteriormente.

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 di autenticazione API.

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

Puoi abilitare o disabilitare l'API REST stabile oppure cambiare l'utente predefinito mediante l'override della seguente configurazione di Airflow opzioni:

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.

Abilita l'API REST Airflow sperimentale

Per impostazione predefinita, la funzionalità di autenticazione dell'API è disattivata nella tramite Google Cloud CLI o tramite l'API Compute Engine. Il server web di Airflow rifiuta tutte le richieste che effettui.

Per abilitare la funzionalità di autenticazione API e l'API sperimentale Airflow 2: Esegui l'override della seguente opzione 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.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 Airflow accetta tutte le API richieste senza autenticazione. Anche se lo stesso server web Airflow non richiede l'autenticazione, è comunque protetto da Identity-Aware Proxy, fornisce il proprio livello di autenticazione.

Consenti le chiamate API all'API REST di Airflow utilizzando il controllo dell'accesso del server web

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

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

Effettua chiamate all'API REST Airflow

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

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

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)

Attenzione: non inserire una barra rovesciata alla fine dell'URL del server web.

Ad esempio, la configurazione seguente non è corretta

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

Accedi all'API REST Airflow utilizzando un account di servizio

Il database Airflow limita la lunghezza del campo email a 64 caratteri. A volte gli account di servizio hanno indirizzi email superiori a 64 caratteri. Non è possibile creare utenti Airflow per questo servizio come di consueto. Se non esiste un utente Airflow per un servizio di questo tipo l'accesso all'API REST Airflow comporta gli errori HTTP 401 e 403.

Come soluzione alternativa, puoi preregistrare un utente Airflow per un account di servizio. A 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 dell'account di servizio.
   • PROJECT_ID con ID progetto.

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

   UI di Airflow

   1. Vai alla UI di Airflow.

   2. Vai a Sicurezza > Elenco utenti e fai clic su Aggiungi un nuovo record. L'utente Airflow deve avere il ruolo Admin per apri questa pagina.

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

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

   5. Specifica il ruolo dell'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 questo 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 dell'utente Airflow. Puoi utilizzare qualsiasi stringa univoca.

3. Dopo aver creato un utente Airflow per un account di servizio, un chiamante l'account di servizio è riconosciuto come utente preregistrato, e ha eseguito l'accesso ad Airflow.

   Nota: per le configurazioni della federazione delle identità per la forza lavoro, tieni presente che sono considerati Account Google, anche se utilizzi la forza lavoro. la federazione delle identità. Quando interagisci con la UI di Airflow tramite un account di servizio, assicurati di utilizzare gli URL dedicati agli Account Google. Vedi Accedere alla UI di Airflow.

Scalabilità del componente API REST Airflow

L'API REST di Airflow e gli endpoint UI di Airflow vengono eseguiti all'interno del componente, ad es. Airflow Webserver. Se utilizzi l'API REST in modo intensivo, considera aumentando i parametri di CPU e memoria per regolare le risorse Airflow Webserver rispetto al carico previsto.

Passaggi successivi

• Attivazione dei DAG con Cloud Functions
• Interfaccia a riga di comando di Airflow