HTTP 요청

requests와 httpx로 REST API 호출, 세션 관리, 인증 패턴을 익힙니다.

---

설치

pip install requests httpx

---

requests — 기본 HTTP 클라이언트

import requests

# GET
response = requests.get("https://api.github.com/users/octocat")
print(response.status_code)   # 200
print(response.json())        # dict
print(response.headers["content-type"])

# POST
data = {"username": "alice", "password": "secret"}
response = requests.post("https://httpbin.org/post", json=data)

# PUT / PATCH / DELETE
requests.put("https://api.example.com/items/1", json={"name": "Updated"})
requests.patch("https://api.example.com/items/1", json={"stock": 5})
requests.delete("https://api.example.com/items/1")

---

요청 파라미터

# Query parameters
response = requests.get(
    "https://api.example.com/products",
    params={"page": 1, "limit": 20, "category": "books"},
)
# → https://api.example.com/products?page=1&limit=20&category=books

# 헤더
headers = {
    "Authorization": "Bearer eyJhbGciOiJIUzI1NiJ9...",
    "Content-Type": "application/json",
    "User-Agent": "MyApp/1.0",
}
response = requests.get("https://api.example.com/me", headers=headers)

# 타임아웃 설정 (필수!)
response = requests.get("https://api.example.com/data",
                        timeout=(5, 30))  # (연결 타임아웃, 읽기 타임아웃)

# 파일 업로드
with open("photo.jpg", "rb") as f:
    response = requests.post(
        "https://api.example.com/upload",
        files={"file": ("photo.jpg", f, "image/jpeg")},
        data={"description": "Profile photo"},
    )

---

Session — 연결 재사용 + 공통 설정

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 재시도 전략
retry_strategy = Retry(
    total=3,               # 최대 3회 재시도
    backoff_factor=1,      # 1s, 2s, 4s 지수 백오프
    status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)

session = requests.Session()
session.mount("https://", adapter)
session.mount("http://", adapter)

# 공통 헤더 설정
session.headers.update({
    "Authorization": "Bearer my_token",
    "Accept": "application/json",
})

# 세션으로 요청 (연결 재사용, 쿠키 유지)
r1 = session.get("https://api.example.com/users")
r2 = session.get("https://api.example.com/products")

session.close()  # 또는 with 문 사용
with requests.Session() as s:
    s.headers.update({"Authorization": "Bearer token"})
    response = s.get("https://api.example.com/data")

---

에러 처리

import requests
from requests.exceptions import (
    Timeout,
    ConnectionError,
    HTTPError,
    RequestException,
)


def safe_request(url: str, **kwargs) -> dict | None:
    try:
        response = requests.get(url, timeout=10, **kwargs)
        response.raise_for_status()  # 4xx, 5xx → HTTPError 발생
        return response.json()
    except Timeout:
        print(f"요청 타임아웃: {url}")
    except ConnectionError:
        print(f"연결 실패: {url}")
    except HTTPError as e:
        print(f"HTTP 오류 {e.response.status_code}: {url}")
        if e.response.status_code == 401:
            print("인증 필요 — 토큰을 확인하세요")
        elif e.response.status_code == 429:
            retry_after = e.response.headers.get("Retry-After", 60)
            print(f"{retry_after}초 후 재시도")
    except RequestException as e:
        print(f"요청 실패: {e}")
    return None

---

httpx — 동기/비동기 통합 클라이언트

import httpx

# 동기 (requests와 유사한 API)
with httpx.Client(timeout=10.0, follow_redirects=True) as client:
    response = client.get("https://api.github.com/users/octocat")
    data = response.json()

# 비동기 (asyncio 기반)
import asyncio


async def fetch_multiple(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=10.0) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses if r.status_code == 200]


# 병렬 API 호출 예제
async def get_user_data(user_ids: list[int]) -> list[dict]:
    async with httpx.AsyncClient(
        base_url="https://api.example.com",
        headers={"Authorization": "Bearer token"},
        timeout=15.0,
    ) as client:
        tasks = [client.get(f"/users/{uid}") for uid in user_ids]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

        results = []
        for uid, resp in zip(user_ids, responses):
            if isinstance(resp, Exception):
                print(f"사용자 {uid} 조회 실패: {resp}")
            elif resp.status_code == 200:
                results.append(resp.json())
        return results


# 실행
user_ids = [1, 2, 3, 4, 5]
users = asyncio.run(get_user_data(user_ids))

---

인증 패턴

import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth


# 1. Basic Auth
response = requests.get(url, auth=HTTPBasicAuth("user", "pass"))
response = requests.get(url, auth=("user", "pass"))  # 단축형

# 2. Bearer Token (JWT)
headers = {"Authorization": f"Bearer {access_token}"}
response = requests.get(url, headers=headers)

# 3. API Key
response = requests.get(url, headers={"X-API-Key": "my_api_key"})
response = requests.get(url, params={"api_key": "my_api_key"})

# 4. OAuth2 (자동 토큰 갱신)
from requests_oauthlib import OAuth2Session  # pip install requests-oauthlib

oauth = OAuth2Session(client_id, token=token)
response = oauth.get("https://api.example.com/resource")

---

REST API 클라이언트 클래스 패턴

import requests
from typing import Any


class APIClient:
    def __init__(self, base_url: str, api_key: str):
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })
        self.base_url = base_url.rstrip("/")

    def _request(self, method: str, path: str, **kwargs) -> Any:
        url = f"{self.base_url}/{path.lstrip('/')}"
        response = self.session.request(method, url, timeout=15, **kwargs)
        response.raise_for_status()
        return response.json() if response.content else None

    def get(self, path: str, params: dict = None): return self._request("GET", path, params=params)
    def post(self, path: str, data: dict): return self._request("POST", path, json=data)
    def put(self, path: str, data: dict): return self._request("PUT", path, json=data)
    def delete(self, path: str): return self._request("DELETE", path)

    def __enter__(self): return self
    def __exit__(self, *args): self.session.close()


# 사용
with APIClient("https://api.example.com", api_key="abc123") as client:
    users = client.get("/users", params={"page": 1})
    new_user = client.post("/users", data={"name": "Alice"})

---

정리

기능	requests	httpx
동기 요청	✅	✅
비동기 요청	❌	✅ (AsyncClient)
HTTP/2	❌	✅
타입 힌트	부분	✅
재시도	HTTPAdapter	httpx.HTTPTransport

requests는 단순한 동기 호출, httpx는 비동기/HTTP2 지원이 필요할 때 선택합니다.