[Airflow] API호출을 위한 Airflow.cfg 파일 수정

api호출 하기 위한 방법

airflow.cfg파일에서

[api]
auth_backends = 

이 부분을 건드려야한다.

1. session

[api]
auth_backends = airflow.api.auth.backend.session

session으로 하는 방법은 web UI에서 쿠키를 찾아 session을 직접 잡아주는 거다.

하지만 이 방법은 로그인 할 때마다 변경되므로 추천 하지 않음.

2. default

[api]
auth_backends = airflow.api.auth.backend.default

이 방식은 인증 비활성화 하는 방법이므로 여러 시스템이 함께 작업한다면 권장하지 않음

3. Kerberos

현재는 지원하지 않음. 하려면 굳이 하려면

[api]
auth_backends = airflow.api.auth.backend.kerberos_auth

[kerberos]
keytab = <KEYTAB>

이 방식을 사용하면 되는데, 굳이 해보진 않았음

4. basic_auth

[api]
auth_backends = airflow.api.auth.backend.basic_auth

기본 인증방식이다. airflow 계정을 사용하면 된다.

ENDPOINT_URL="http://localhost:8080/"
curl -X GET\    --user "username:password"\    "${ENDPOINT_URL}/api/v1/pools"

curl -X GET  \

    --user "user:pwd" \

    "http://10.182.23.157:8080/api/v1/dags"

이런 식으로 사용하면 된다.

파이썬으로 호출하는 방법은 아래와 같다.

import requests
from requests.auth import HTTPBasicAuth

# API URL 및 사용자 인증 정보 설정
url = "http://10.182.23.157:8080/api/v1/dags"
username = "admin"
password = "admin"

response = requests.get(url, auth=HTTPBasicAuth(username, password))

print(f"Status Code: {response.status_code}")
if response.status_code == 200:
    # 응답이 성공적일 경우 JSON 데이터 출력
    data = response.json()
    print(data)
else:
    print(f"Failed to fetch DAGs. Status Code: {response.status_code}")

다음은 Owner명으로 구분하는 방법이다

import requests
from requests.auth import HTTPBasicAuth
import os

# API URL 및 사용자 인증 정보 설정
url = "http://10.182.23.157:8080/api/v1/dags"
username = "user"   #로그인 ID
password = "pwd"    #로그인 PWD

response = requests.get(url, auth=HTTPBasicAuth(username, password))

if response.status_code == 200:
    data = response.json()
    if "dags" in data:
        file_names = []
        for dag in data["dags"]:
            owners = dag.get("owners", [])
            if "OWNER명" in owners:  # owner명을 설정
                file_path = dag.get("fileloc", "")
                file_name = os.path.basename(file_path)  # 파일명 추출
                file_names.append(file_name)
        print("파일명 :", file_names)
    else:
        print("owner 없음.")
else:
    print(f"error: {response.status_code}")