Airflow HTTP 센서로 AI 영상 Job 완료 기다리기

Airflow HTTP 센서(HttpSensor)는 poke할 때마다 엔드포인트에 GET을 보내고, response_check 콜러블이 True를 반환하면 성공합니다. AI 영상 Job을 기다리려면 HttpOperator로 Job을 제출한 뒤, Job이 끝나면 True를 반환하는 response_check를 단 HttpSensor를 Job의 상태 URL에 연결하세요. 모드는 reschedule로 두고, 센서 자체의 timeout도 설정하세요.

Airflow 관련 내용은 HTTP 프로바이더의 오퍼레이터 가이드와 HttpSensor, HttpOperator, 커넥션 페이지, 그리고 Airflow의 센서, 태스크, Task SDK 레퍼런스에서, Sume 관련 내용은 영상 생성 (영문), Job과 결과 (영문)에서 가져왔습니다. 모두 2026-09-27에 확인했습니다. Sume에는 Airflow 프로바이더가 없으며, DAG가 일반 HTTPS 호출을 보냅니다. Airflow 밖에서 쓰는 폴링 루프는 영상 생성 Job 상태 API 폴링하기에서 다룹니다.

HTTP 오퍼레이터로 Job은 어떻게 시작하나요?

HTTP 커넥션(여기서는 sume_api)을 만들고 호스트는 api.sume.com, 스키마는 https로 설정하세요. Airflow의 HTTP 오퍼레이터는 기본값이 http입니다. Airflow는 커넥션의 Extra에서 헤더를 JSON으로 받으므로, Authorization: Bearer <your key>는 그곳에 넣고 키는 DAG 코드 밖에 두세요.

HttpOperator는 기본적으로 POST를 보내며, endpoint, data, headers는 템플릿으로 처리됩니다. Sume의 POST /v1/videos는 id, polling_url, status: "pending", model과 함께 202로 응답합니다. id를 반환하는 response_filter를 두면 Job ID가 태스크의 결과가 되고, 센서는 이 값을 XCom에서 가져옵니다.

Idempotency-Key: video-{{ run_id }}로 요청을 DAG 실행에 묶으세요. Airflow는 DAG 안에서 고유한 값이 필요하면 logical date 대신 run_id를 쓰라고 안내하며, Sume는 다시 보낸 키에 두 번째 유료 Job 대신 원래 Job으로 응답합니다. 이렇게 하면 태스크 재시도뿐 아니라 deferrable 모드에도 대비할 수 있습니다. Airflow는 deferrable 모드에서 triggerer가 재시작되면 POST가 다시 실행될 수 있다고 경고합니다.

# Inside your DAG. The sume_api connection holds the host and the key.
import json
from airflow.providers.http.operators.http import HttpOperator
from airflow.providers.http.sensors.http import HttpSensor

def video_done(response):
    status = response.json()["status"]
    if status in ("failed", "cancelled"):
        raise ValueError(f"Sume video job ended as {status}")
    return status == "completed"

submit = HttpOperator(
    task_id="submit_video", http_conn_id="sume_api", endpoint="v1/videos",
    data=json.dumps({"model": "sume/auto", "prompt": "A slow pan across a desk, morning light",
                     "aspect_ratio": "9:16", "duration": 5}),
    headers={"Content-Type": "application/json", "Idempotency-Key": "video-{{ run_id }}"},
    response_filter=lambda response: response.json()["id"],
)
wait = HttpSensor(
    task_id="wait_for_video", http_conn_id="sume_api",
    endpoint="v1/videos/{{ ti.xcom_pull(task_ids='submit_video') }}",
    response_check=video_done, response_error_codes_allowlist=["429"],
    mode="reschedule", poke_interval=90, timeout=20 * 60,
)
submit >> wait

Job 상태별로 response_check는 무엇을 반환해야 하나요?

True면 센서가 통과하고, False면 다시 poke합니다. 두 가지 실패 상태에서는 예외를 던지세요. 예외는 센서 태스크를 실패시키고, 그러면 태스크는 타임아웃까지 poke를 계속하는 대신 retries 설정을 따릅니다.

response_error_codes_allowlist는 왜 설정하나요?

기본적으로 HttpSensor는 404에서만 False를 반환합니다. 다른 오류 코드는 모두 예외를 일으켜 센서 자체를 실패시키고, poke도 더는 하지 않습니다. Sume 폴링은 키의 읽기 예산에서 차감되며, 예산을 다 쓰면 429가 돌아옵니다. Sume 문서는 이를 Job의 결과로 다루지 말고 백오프하라고 안내합니다. 허용 목록에 ["429"]를 넣으면 요청 한도에 걸려도 태스크 대신 poke 한 번만 잃습니다. 이 목록은 기본값 ["404"]를 대체하므로, 이제 Sume가 모르는 Job ID는 타임아웃까지 poke하는 대신 센서를 실패시킵니다. 읽기 예산은 쓰기 예산과 별개이고 그 마흔 배 크기이므로, 폴링 때문에 제출이 막히지는 않습니다.

센서의 timeout과 poke 간격은 얼마로 잡아야 하나요?

mode="reschedule"을 쓰세요. 기본 poke 모드에서는 센서가 실행되는 내내 워커 슬롯을 차지하지만, reschedule 모드에서는 확인하는 동안에만 슬롯을 씁니다. Airflow 레퍼런스는 스케줄러의 부담을 덜기 위해 reschedule 모드의 poke 간격을 일 분보다 길게 잡으라고 하고, Sume 문서는 적절한 폴링 간격으로 30초를 제시하면서 영상은 보통 30초에서 몇 분이 걸린다고 설명하므로, 90초가 양쪽에 모두 맞습니다.

timeout은 첫 poke부터 reschedule 대기 시간까지 포함해 계산되며, 이를 넘기면 AirflowSensorTimeout이 발생해 재시도 없이 센서가 즉시 실패합니다. Sume 문서는 영상의 클라이언트 쪽 마감 시간으로 20분이 적당하다고 봅니다. 센서 타임아웃은 Job을 취소하지 않습니다. Job은 계속 실행되고 계속 과금됩니다. 새로 제출하지 말고 같은 Job ID를 다시 확인하세요. 같은 DAG 실행 안에서 submit_video를 다시 돌려도 같은 키를 보내므로 원래 Job이 돌아옵니다.

• HttpSensor는 영상을 다운로드하지 않습니다. 완료된 Job에는 unsigned_urls가 나열되고, 키를 붙여 GET /v1/videos/{id}/content를 호출하면 파일로 리다이렉트됩니다. Sume API에서 생성한 영상 다운로드하기를 참고하세요.
• HttpSensor는 deferrable=True로 deferrable 모드에서도 실행할 수 있습니다.