httpx AsyncClient로 외부 API 비동기 호출하기

여러 외부 API를 호출하는 서버라면, 요청을 기다리는 동안 다른 일을 처리하는 비동기 방식이 유리합니다. 파이썬에서 비동기 HTTP 호출에 널리 쓰이는 httpx의 AsyncClient 사용법을, 실제 운영 중인 수집·발행 코드에서 쓰는 형태를 기준으로 정리합니다.

이 글의 가정: async/await(파이썬 비동기 문법) 기본 개념을 알고 있고, Python 3.8 이상 환경을 가정합니다. 설치는 pip install httpx입니다.

AsyncClient 기본 사용

httpx는 동기 방식과 비동기 방식을 모두 지원합니다. 비동기에서는 AsyncClientasync with 블록으로 열고 await로 요청을 보냅니다.

import httpx

async with httpx.AsyncClient(timeout=30) as client:
    resp = await client.post(url, headers=headers, json=payload)
    data = resp.json()

async with로 열면 블록이 끝날 때 클라이언트가 자동으로 정리됩니다.

타임아웃 설정

외부 API가 응답하지 않을 때 무한정 기다리지 않도록 timeout을 지정하는 것이 중요합니다. 호출의 성격에 따라 값을 다르게 둘 수 있습니다. 예를 들어 인증 토큰 발급처럼 빨라야 하는 호출은 약 10초, 데이터 수집은 약 30초, 생성 시간이 긴 AI 언어 모델 호출은 약 40초처럼 맥락에 맞춰 정합니다.

응답 처리와 에러 스킵

여러 요청을 다룰 때는, 하나가 실패해도 전체가 멈추지 않도록 상태 코드를 확인하고 건너뛰는 방식이 실용적입니다.

async with httpx.AsyncClient(timeout=30) as client:
    resp = await client.get(api_url, params={"q": keyword})
    if resp.status_code != 200:
        print(f"오류: {resp.status_code}")
        # 이 항목은 건너뛰기
    else:
        data = resp.json()

여러 API를 순회 호출하기

키워드 목록처럼 여러 대상을 도는 경우, 실패한 항목만 로그를 남기고 continue로 넘어갑니다.

for keyword in keywords:
    async with httpx.AsyncClient(timeout=30) as client:
        res = await client.get(api_url, params={"q": keyword})
        if res.status_code != 200:
            print(f"'{keyword}' 오류: {res.status_code}")
            continue
        results.append(res.json())

이 방식은 구현이 단순하다는 장점이 있지만, 실패한 요청을 다시 시도하지는 않습니다. 즉 일시적 오류도 그냥 건너뜁니다. 실패를 놓치면 안 되는 호출에는 재시도를 덧붙이는 편이 좋습니다.

선택적: 재시도 덧붙이기

다시 시도해도 소용없는 오류(400, 401, 403, 404)는 즉시 멈추고, 서버 일시 오류나 타임아웃만 몇 번 재시도하는 패턴입니다.

NO_RETRY = {400, 401, 403, 404}

async def call_with_retry(make_request, tries=3):
    for _ in range(tries):
        try:
            return await make_request()
        except httpx.HTTPStatusError as e:
            if e.response.status_code in NO_RETRY:
                raise
        except httpx.TimeoutException:
            pass
    raise RuntimeError("재시도 소진")

호출별로 이 래퍼를 씌우면, 단순 스킵보다 견고하게 일시적 오류를 흡수할 수 있습니다.

마무리

정리하면 AsyncClient + async with + timeout이 비동기 호출의 기본이고, 여기에 상태 코드 확인과 (필요하다면) 재시도를 더하면 됩니다. 이 패턴은 주기 작업이 부르는 수집기와 발행기 양쪽에서 똑같이 쓰입니다. 스케줄러가 이 호출을 어떻게 주기적으로 돌리는지는 FastAPI에서 APScheduler로 주기 작업 구현하기, 같은 패턴으로 글을 올리는 예는 Blogger API와 OAuth 2.0으로 블로그 글 자동 발행하기에서 볼 수 있습니다. API 호출량을 효율적으로 다루는 방법은 YouTube Data API 쿼터 초과를 피하는 방법을 참고하세요.

이 글이 도움이 됐다면 구독과 공유 부탁드립니다.

댓글

이 블로그의 인기 게시물

UGREEN DXP4800PLUS NAS로 유튜브 쇼츠 자동화 시스템 구축기 (Docker · FastAPI · PostgreSQL · Edge-TTS)

YouTube Data API 쿼터 초과를 피하는 방법 (700개 쇼츠 수집 사례)

FastAPI app.mount 사용 후 API가 404가 되는 이유와 해결 방법