Python 비동기 HTTP 클라이언트

import httpx
import asyncio
import logging
from typing import Optional, Any, Dict

logger = logging.getLogger(__name__)

class AsyncHttpClient:
    """httpx 기반의 비동기 HTTP 클라이언트 래퍼입니다."""
    
    def __init__(self, base_url: str, timeout: float = 10.0):
        self.base_url = base_url
        self.timeout = timeout
        self.client: Optional[httpx.AsyncClient] = None

    async def __aenter__(self):
        self.client = httpx.AsyncClient(
            base_url=self.base_url,
            timeout=self.timeout,
            headers={"Content-Type": "application/json"}
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.client:
            await self.client.aclose()

    async def request(
        self,
        method: str,
        path: str,
        data: Optional[Dict[str, Any]] = None,
        params: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """일반화된 요청 메서드입니다."""
        if not self.client:
            raise RuntimeError("Client is not initialized. Use 'async with'.")

        try:
            response = await self.client.request(
                method, path, json=data, params=params
            )
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            logger.error(f"HTTP error occurred: {e.response.status_code} - {e.response.text}")
            throw e
        except Exception as e:
            logger.error(f"An unexpected error occurred: {e}")
            throw e

    async def get(self, path: str, params: Optional[Dict[str, Any]] = None):
        return await self.request("GET", path, params=params)

    async def post(self, path: str, data: Dict[str, Any]):
        return await self.request("POST", path, data=data)

# 사용 예시
async def main():
    async with AsyncHttpClient("https://jsonplaceholder.typicode.com") as client:
        # 단일 요청
        post = await client.get("/posts/1")
        print(f"Post Title: {post['title']}")

        # 병렬 요청
        tasks = [client.get(f"/posts/{i}") for i in range(1, 6)]
        results = await asyncio.gather(*tasks)
        print(f"Fetched {len(results)} posts")

if __name__ == "__main__":
    asyncio.run(main())