aiohttp 비동기로 API 호출하기

mingkeNov 23, 2023

소개

FastAPI로 개발하면서 외부로 API호출 할 때 httpx 라이브러리를 많이 사용합니다. httpx는 동기, 비동기 방식을 모두 지원하고 공식문서에서도 httpx를 사용하고 있습니다.

하지만 오늘은 aiohttp를 이용해서 비동기 API호출 방법에 대해서 알아보겠습니다.

aiohttp란

Python의 비동기 HTTP 클라이언트/서버 프레임워크입니다. 이것은 Python의 비동기 프로그래밍 기능을 활용하여, 동시에 여러 HTTP 요청을 처리할 수 있게 해주는 강력한 도구입니다. 동시에 여러 네트워크 요청을 처리할 때 I/O 대기 시간을 효율적으로 관리할 수 있게 해줍니다.

클라이언트와 서버사이드 양쪽의 기능을 모두 제공하지만 오늘은 클라이언트 사이드에서 비동기 API호출 방식에 대해서만 알아보겠습니다. 서버는 굳이 aiohttp를 사용할 필요가 없을 것 같습니다.

구현

기본적으로 다음과 같이 사용할 수 있습니다. with문을 사용해서 컨텍스트안에서 세션을 만들어 사용할 수 있습니다.

import aiohttp
 
async def request():
    try:
        async with aiohttp.ClientSession() as session:
            URI = "https://example.com"
                headers = {"example": "example"}
            async with session.post(URI, headers=headers) as res:
                response = await res.json()
                return response
    except aiohttp.ClientError as e:
        raise e

위 방식대로 하면 request를 여러번 실행했을 때 session을 계속 열었다 닫았다 해야해서 저는 다음과 같이 많이 사용합니다. 최근에 FastAPI를 많이 쓰기 때문에 FastAPI에서 lifespan에 api_client를 주입해서 사용할 수도 있습니다.

import aiohttp
import warnings
from typing import Callable, Any, Coroutine, TypeVar
from functools import wraps
 
T = TypeVar("T", bound="APIClient")
 
# 세션이 실행되지 않았을 때 에러를 발생시키는 데코레이터
def ensure_session(
    func: Callable[..., Coroutine[Any, Any, aiohttp.ClientResponse]]
) -> Callable[..., Coroutine[Any, Any, aiohttp.ClientResponse]]:
    @wraps(func)
    async def wrapper(self: T, *args: Any, **kwargs: Any) -> aiohttp.ClientResponse:
        if self.session is None:
            raise RuntimeError("세션이 시작되지 않았습니다. start() 메소드를 호출하세요.")
        return await func(self, *args, **kwargs)
 
    return wrapper
 
class APIClient:
    """
    APIClient객체를 단독적으로 사용할 수도 있고 다른 클래스에 주입하여 사용할 수 도 있음
    """
 
    def __init__(self) -> None:
        self.session = None
 
    async def start(self) -> None:
        if not self.session:
            self.session = aiohttp.ClientSession()
 
    async def close(self) -> None:
        if self.session:
            await self.session.close()
            self.session = None
 
    @ensure_session
    async def get(
        self, url: str, params: dict[str, Any] | None = None, **kwargs: Any
    ) -> aiohttp.ClientResponse:
        response = await self.session.get(url, params=params, **kwargs)
        return response
    ...
    # 같은 방식으로 다른 메소드 구현
 
    def __del__(self) -> None:
        if self.session and not self.session.closed:
            warnings.warn("APIClient 객체가 close되지 않았습니다.")

async def test():
    api_client = APIClient()
    await api_client.start()
    res1 = await api_client.get("https://www.naver.com")
    res2 = await api_client.get("https://www.naver.com")
    await api_client.close()
    return res1, res2
 
res_1, res_2 = await test()
 
class Test:
	"""
    some_method를 시작하고 닫을 때 start와 close를 반드시 실행
    """
    def __init__(self) -> None:
        self.api_client = APIClient()
 
    async def some_method(self, ...):
        response = await api_client.get("https://www.google.com")
    ...

오늘은 간단하게 aiohttp client로 API호출하는 방법을 알아봤습니다. 비동기 처리 능력은 현대 웹 애플리케이션 개발에 있어 중요한 요소이기 때문에 알아두면 좋다고 생각합니다. 이상으로 간략히 알아본 aiohttp 비동기로 API 호출하기 포스팅을 마치겠습니다.

목차

소개

aiohttp란

구현