Mengakses Airflow REST API

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Apache Airflow memiliki antarmuka REST API yang dapat Anda gunakan untuk melakukan tugas seperti mendapatkan informasi tentang tugas dan eksekusi DAG, memperbarui DAG, mendapatkan konfigurasi Airflow, menambahkan dan menghapus koneksi, serta mencantumkan pengguna.

Untuk contoh penggunaan Airflow REST API dengan fungsi Cloud Run, lihat Memicu DAG dengan fungsi Cloud Run.

Versi Airflow REST API

• Airflow 3 menggunakan Airflow REST API v2.
• Airflow 2 menggunakan Airflow REST API v1.

Mengonfigurasi Airflow REST API

Airflow 3

Airflow 3 menggunakan Airflow REST API v2.

Cloud Composer menggunakan backend autentikasi API-nya sendiri.

Otorisasi berfungsi dengan cara standar yang disediakan oleh Airflow 3. Saat pengguna baru memberikan otorisasi melalui API, akun pengguna akan mendapatkan peran Op secara default.

Airflow REST API di Airflow 3 selalu diaktifkan dan tidak dapat dinonaktifkan. Anda dapat mengubah peran pengguna default dengan mengganti opsi konfigurasi Airflow berikut:

Bagian	Kunci	Nilai	Catatan
api	composer_auth_user_registration_role	Op	Anda dapat menentukan peran lainnya.

Airflow 2

Airflow REST API v1 diaktifkan secara default di Airflow 2.

Cloud Composer menggunakan backend autentikasi API-nya sendiri.

Otorisasi berfungsi dengan cara standar yang disediakan oleh Airflow. Saat pengguna baru memberikan otorisasi melalui API, akun pengguna akan mendapatkan peran Op secara default.

Anda dapat mengaktifkan atau menonaktifkan Airflow REST API, atau mengubah peran pengguna default dengan mengganti opsi konfigurasi Airflow berikut:

Bagian	Kunci	Nilai	Catatan
api	auth_backends	airflow.composer.api.backend.composer_auth	Untuk menonaktifkan Airflow REST API, ubah ke airflow.api.auth.backend.deny_all
api	composer_auth_user_registration_role	Op	Anda dapat menentukan peran lainnya.

Mengizinkan panggilan API ke Airflow REST API menggunakan kontrol akses server web

Bergantung pada metode yang digunakan untuk memanggil Airflow REST API, metode pemanggil dapat menggunakan alamat IPv4 atau IPv6. Jangan lupa untuk membuka blokir traffic IP ke Airflow REST API menggunakan kontrol akses server web.

Gunakan opsi konfigurasi default, yaitu All IP addresses have access (default) jika Anda tidak yakin dari alamat IP mana panggilan ke Airflow REST API akan dikirim.

Melakukan panggilan ke Airflow REST API

Bagian ini memberikan contoh skrip Python yang dapat Anda gunakan untuk memicu DAG dengan Airflow REST API.

Dalam skrip, tetapkan variabel berikut:

• dag_id: nama DAG, seperti yang ditentukan dalam file sumber DAG.
• dag_config: konfigurasi untuk operasi DAG.

• web_server_url: URL server web Airflow Anda. Formatnya adalah https://<web-server-id>.composer.googleusercontent.com.

  Penting: Jangan letakkan garis miring terbalik di akhir URL server web.

• (Airflow 3) logical_date: Tanggal logis untuk eksekusi 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)

Mengakses Airflow REST API menggunakan akun layanan

Database Airflow di Airflow versi sebelum 2.3.0 membatasi panjang kolom email hingga 64 karakter. Akun layanan terkadang memiliki alamat email yang lebih panjang dari 64 karakter. Pengguna Airflow tidak dapat dibuat untuk akun layanan tersebut dengan cara biasa. Jika tidak ada pengguna Airflow untuk akun layanan tersebut, mengakses Airflow REST API akan menghasilkan error HTTP 401 dan 403.

Catatan: Error 401 (Tidak diotorisasi) dan 403 (Dilarang) adalah error HTTP umum. Jika Anda masih mendapatkan error ini setelah menggunakan solusi sementara yang dijelaskan, periksa kemungkinan penyebab lainnya. Misalnya, peran IAM mungkin tidak ada di akun layanan. Untuk mengizinkan akun layanan mengakses Airflow REST API, berikan peran IAM yang mengizinkan akses ke antarmuka web.

Sebagai solusi sementara, Anda dapat melakukan pra-pendaftaran pengguna Airflow untuk akun layanan. Untuk melakukannya, gunakan accounts.google.com:NUMERIC_USER_ID sebagai nama pengguna, dan string unik apa pun sebagai email.

1. Untuk mendapatkan NUMERIC_USER_ID untuk akun layanan, jalankan:

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

   Ganti:

   • SA_NAME dengan nama akun layanan.
   • PROJECT_ID dengan Project ID.

2. Buat pengguna Airflow dengan peran Op untuk akun layanan:

   UI Airflow

   1. Buka UI Airflow.

   2. Buka Keamanan > Buat Daftar Pengguna, lalu klik Tambahkan data baru. Pengguna Airflow Anda harus memiliki peran Admin untuk membuka halaman ini.

   3. Tentukan accounts.google.com:NUMERIC_USER_ID sebagai nama pengguna. Ganti NUMERIC_USER_ID dengan ID pengguna yang diperoleh pada langkah sebelumnya.

   4. Tentukan ID unik sebagai email. Anda dapat menggunakan string unik apa pun.

   5. Tentukan peran untuk pengguna. Misalnya, Op.

   6. Pastikan kotak centang Aktif? dicentang.

   7. Tentukan nama depan dan nama belakang pengguna. Anda dapat menggunakan string apa pun.

   8. Klik Simpan.

   gcloud

   Jalankan perintah Airflow CLI berikut:

   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
   

   Ganti:

   • ENVIRONMENT_NAME dengan nama lingkungan.
   • LOCATION dengan region tempat lingkungan berada.
   • NUMERIC_USER_ID dengan ID pengguna yang diperoleh di langkah sebelumnya.
   • UNIQUE_ID dengan ID untuk pengguna Airflow. Anda dapat menggunakan string unik apa pun.

3. Setelah Anda membuat pengguna Airflow untuk akun layanan, pemanggil yang diautentikasi sebagai akun layanan akan dikenali sebagai pengguna yang telah terdaftar sebelumnya, dan akan login ke Airflow.

Menskalakan komponen Airflow REST API

Endpoint Airflow REST API dan Airflow UI dijalankan dalam server web Airflow. Jika Anda menggunakan REST API secara intensif, pertimbangkan untuk meningkatkan jumlah CPU dan memori yang tersedia untuk server web Airflow, berdasarkan perkiraan beban.

Langkah berikutnya

• Memicu DAG dengan Cloud Run Functions
• Antarmuka command line Airflow