HTTPX Snippets: 실용 가이드

1 HTTPX 소개

HTTPX는 Python을 위한 현대적인 HTTP 클라이언트입니다. familiar requests API를 유지하면서 HTTP/2, async 지원, 연결 풀링을 기본적으로 제공합니다. requests를 알고 있다면 HTTPX의 대부분을 이미 알고 있는 것입니다. 단지 머릿속으로 "동기 또는 비동기?"를 체크하는 것만 추가하면 됩니다.

동기 및 비동기 동일한 정신 모델로
현대 프로토콜: 가능한 경우 HTTP/2
좋은 기본값: 연결 풀링, 타임아웃, 재시도

"다음"을 클릭하여 HTTPX를 설치하고 첫 번째 요청을 보내세요.

2 설치

가상 환경을 사용한 다음 HTTPX를 설치하세요. [http] 추가 기능은 표준 전송을 가져옵니다 (추가 종속성 불필요).

pip install httpx[http]

HTTP/2가 필요한 경우 h2를 설치하세요: pip install h2. macOS에서 HTTPS의 경우 인증서가 필요할 수 있습니다: pip install certifi 및 SSL_CERT_FILE 또는 REQUESTS_CA_BUNDLE을 설정하세요.

3 간단한 GET

다음은 최소한의 동기 GET입니다. 상태 코드와 짧은 발췌문을 출력합니다.

import httpx

with httpx.Client() as client:
    r = client.get("https://httpbin.org/get")
    print("status:", r.status_code)
    print("content[:120]:", r.text[:120])

클라이언트는 컨텍스트 관리자로 사용되므로 연결이 깔끔하게 해제됩니다. 이것만으로도 "파일 열기 초과" 오류의 놀라울 정도로 많은 수를 방지할 수 있습니다.

4 헤더, 매개변수 및 타임아웃

헤더와 쿼리 매개변수를 일반 딕셔너리로 전달합니다. 항상 명시적인 타임아웃을 설정하세요.

import httpx

with httpx.Client(
    base_url="https://httpbin.org",
    headers={"User-Agent": "httpx-snippets/1.0"},
    params={"page": "1", "per_page": "5"},
    timeout=5.0,
) as client:
    r = client.get("/get")
    r.raise_for_status()
    data = r.json()
    print(data.get("args"))

base_url을 사용하여 중복을 줄이고 r.raise_for_status()를 사용하여 HTTP 오류를 조용히 실패하는 대신 예외로 변환하세요.

5 JSON POST

JSON POST는 한 줄로 처리됩니다. 클라이언트는 직렬화하고 올바른 Content-Type을 설정합니다.

import httpx

payload = {"name": "Ada", "skills": ["math", "coding"]}

with httpx.Client(base_url="https://httpbin.org") as client:
    r = client.post("/post", json=payload)
    r.raise_for_status()
    print(r.json()["json"])

6 오류 처리

가능하면 특정 예외를 사용하고 그렇지 않으면 기본 클래스로 대체하세요.

import httpx
from httpx import ConnectTimeout, ReadTimeout, RemoteProtocolError

try:
    with httpx.Client(timeout=1.0) as client:
        r = client.get("https://httpbin.org/delay/2")
        r.raise_for_status()
except ConnectTimeout:
    print("Could not connect (network down, DNS, refused).")
except ReadTimeout:
    print("Server took too long to respond.")
except RemoteProtocolError:
    print("Connection dropped mid-flight.")
except httpx.HTTPStatusError as e:
    print(f"HTTP error: {e.response.status_code} {e.response.reason_phrase}")
except httpx.RequestError as e:
    print(f"Request failed: {e}")

7 비동기 요청 (asyncio)

동시성이 필요한 경우 httpx.AsyncClient로 전환하세요. asyncio.gather를 사용하여 I/O 호출을 그룹화하세요.

import asyncio
import httpx

async def fetch(url: str):
    async with httpx.AsyncClient(timeout=5.0) as client:
        r = await client.get(url)
        r.raise_for_status()
        return r.json()

async def main():
    urls = ["https://httpbin.org/get"] * 4
    results = await asyncio.gather(*(fetch(u) for u in urls))
    for i, data in enumerate(results, 1):
        print(i, data.get("origin"))

asyncio.run(main())

8 재시도 및 백오프

일시적인 오류의 경우 httpx.Limits 및 Timeout을 사용하여 전송을 재시도 정책으로 래핑하세요.

import httpx
from urllib3.util.retry import Retry
from httpx import LegacyHTTPTransport

retry = Retry(
    total=3,
    backoff_factor=0.5,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS"],
)

limits = httpx.Limits(max_keepalive_connections=5, max_connections=10)
timeout = httpx.Timeout(5.0, connect=2.0)

transport = LegacyHTTPTransport(retries=retry)
client = httpx.Client(transport=transport, limits=limits, timeout=timeout)

r = client.get("https://httpbin.org/status/503")
print(r.status_code)

참고: 재시도 로직은 전송 계층에 있습니다. 보수적으로 유지하세요. 재시도는 중단을 증폭시킬 수 있습니다.

9 고급 타임아웃

연결 및 읽기 타임아웃을 분리하여 더 엄격하게 제어합니다. 스트리밍 응답의 경우 읽기 타임아웃을 비활성화(None)할 수도 있습니다.

import httpx

timeout = httpx.Timeout(connect=2.0, read=5.0, write=2.0, pool=1.0)
with httpx.Client(timeout=timeout) as client:
    r = client.get("https://httpbin.org/stream/3")  # stream lines
    for line in r.iter_lines():
        print(line)

10 파일 업로드 (Multipart)

files=로 파일을 보내고 data=로 일반 폼 필드를 보냅니다.

import httpx

with httpx.Client(base_url="https://httpbin.org") as client:
    with open("README.md", "rb") as f:
        files = {"file": ("README.md", f, "text/markdown")}
        data = {"uploader": "me"}
        r = client.post("/post", files=files, data=data)
        r.raise_for_status()
        print(r.json()["headers"].get("Content-Type"))

11 FastAPI에서 HTTPX 사용

앱 시작 시 단일 재사용 가능한 클라이언트를 생성하고 종료 시 닫으세요.

from fastapi import FastAPI
import httpx

app = FastAPI()
client = httpx.Client(base_url="https://httpbin.org", timeout=5.0)

@app.on_event("startup")
async def on_startup():
    await client.__aenter__()

@app.on_event("shutdown")
async def on_shutdown():
    await client.__aexit__(None, None, None)

@app.get("/proxy/get")
async def proxy_get():
    r = await client.get("/get")
    r.raise_for_status()
    return r.json()

운영 환경에서는 Limits와 재시도 정책을 전송 계층에 마운트하세요. 상위 서비스가 너무 말이 많으면 일반적인 응답을 캐시하세요.