Python aiocache 로 비동기 Slack API 요청 캐싱하기

슬랙 API를 사용하면서 동일한 요청을 반복적으로 해야 할 때가 있습니다. 예를 들어 슬랙의 특정 메시지를 조회하여 데이터를 가져오는 경우가 있을 수 있죠.

 

하지만 슬랙 API는 사용량 제한이 있기 때문에, 같은 요청을 반복하는 대신 캐싱을 통해 효율적으로 처리할 수 있습니다. 오늘은 aiocache 라이브러리를 사용해 비동기 함수에 대한 캐시 구현 방법을 소개해 보겠습니다.

 

 

문제 상황

slack_sdk 에서 제공하는 AsyncWebClient 를 이용하여 특정 메시지의 댓글을 가져오겠습니다. 이를 위해서는 conversations_replies 메서드를 사용합니다.

 

async def fetch_messages(
    client: AsyncWebClient,
    channel_id: str,
    ts: str,
) -> list[dict[str, Any]]:
    res = await client.conversations_replies(channel=channel_id, ts=ts)
    messages = res.get("messages", [])
    return messages

 

특별히 문제가 되는 부분은 없습니다. 하지만 슬랙은 각 API 별로 티어를 구분하여 요청에 제한을 두고 있습니다. 슬랙 봇을 개발하고 있다면 이 부분을 특히 주의해야 합니다.

 

위 예제의 conversations_replies 의 경우, Rate limits 이 Tier 3 입니다. 

 

 

 

위 공식문서에 나온 것처럼 Tier 3 은 분당 50번 이상의 요청을 허용하지만 제가 직접 동시 요청 테스트를 해본 결과 100회를 넘기고 얼마 안 가 ratelimited 에러 응답을 받았습니다. 

 

만약 이 함수가 필요하지 않은 기능이라면 좋겠지만 꼭 호출이 필요하다면? 이 문제를 어떻게 해결할 수 있을까요? 

 

 

aiocache란?

aiocache는 비동기적으로 캐시를 관리할 수 있는 Python 라이브러리입니다. Python에서 기본적으로 제공하는 lru_cache 와 다르게 비동기 함수에 캐시를 쉽게 적용할 수 있는 데코레이터를 제공합니다.

 

캐싱을 사용하면 동일한 요청에 대해 반복 횟수를 줄이고 미리 기억해 둔 값을 빠르게 반환할 수 있습니다. 즉, API 요청은 줄이고 응답 성능은 높이는 거죠.

 

 

 

aiocache 설치하기

먼저 aiocache를 설치해야겠죠? 아래 명령어로 간단하게 설치할 수 있습니다.

pip install aiocache

 

 

 

비동기 함수 캐싱

처음 예제 코드를 다시 가져와보겠습니다.

async def fetch_messages(
    client: AsyncWebClient,
    channel_id: str,
    ts: str,
) -> list[dict[str, Any]]:
    res = await client.conversations_replies(channel=channel_id, ts=ts)
    messages = res.get("messages", [])
    return messages

 

이 코드에 캐싱을 적용하기 위하여 먼저 cache_key_builder 함수를 정의하겠습니다.

 

cache_key_builder 함수

def cache_key_builder(func, *args, **kwargs):
    # `args`에서 `client`를 제외하고 `channel_id`와 `ts`만 사용해 키를 생성
    if "channel_id" in kwargs and "ts" in kwargs:
        channel_id = kwargs["channel_id"]
        ts = kwargs["ts"]
    else:
        # 위치 인자를 사용할 때 `args`에서 두 번째와 세 번째 인자 사용
        channel_id = args[1]
        ts = args[2]
    return f"{func.__name__}:{channel_id}:{ts}"

 

cache_key_builder 함수에서는 캐시 키를 생성할 건데요. 여기서 key_builder는 client를 제외하고, channel_id와 ts만을 조합해 캐시 키를 생성합니다.

 

여기서 client 를 제외하는 이유는 client 객체가 주입될 때 동일한 객체가 주입되는 것인 아닌 새롭게 정의된 client 가 주입되기 때문입니다. 캐싱을 위해서는 동일한 파라미터 값을 키로 가져야 하므로 client 는 제외합니다.

 

 

cache_key_builder 함수를 만들었다면 다음처럼 데코레이터를 작성합니다.

 

@cached 데코레이터 적용

from aiocache import cached


@cached(ttl=60, key_builder=cache_key_builder)
async def fetch_messages(
    client: AsyncWebClient,
    channel_id: str,
    ts: str,
) -> list[dict[str, Any]]:
    res = await client.conversations_replies(channel=channel_id, ts=ts)
    messages = res.get("messages", [])
    return messages

 

 

@cached 데코레이터에서 ttl=60은 캐시 된 값이 60초 동안 유효함을 의미합니다. 그리고 두번째 key_builder 인자에 아까 정의해둔 cache_key_builder 함수를 넣습니다.

이제 fetch_messages 함수는 최초 1회 호출 된 후에 60초 동안 다시 호출되지 않고 캐시된 결과를 반환하게 됩니다. 결과적으로 동일한 channel_id 와 ts 라면 분당 1회만 요청하는 것이죠.

 

이렇게 비동기 함수에 대한 캐싱이 끝났습니다. 생각보다 간단하죠? 😉

 

 

aiocache 와 lru_cache 비교

aiocache와 functools.lru_cache는 둘 다 Python에서 캐싱을 위해 사용하지만 각각의 특징과 차이점이 있습니다.

 

첫 번째 차이는 aiocache 는 비동기 함수를 캐싱하는 데 최적화되어 있지만, lru_cache 는 비동기 함수에 부적합합니다.

lru_cache 가 비동기 함수에는 적절하지 않다는 공식문서 내용.

 

 

두 번째 차이는 aiocache 는 다양한 캐시 백엔드를 지원합니다. (예: SimpleMemoryCache, RedisCache, MemcachedCache), 이와 달리 lru_cache 는 단일 프로세스 내 메모리만 사용합니다. 프로세스가 종료되면 데이터가 사라지죠.

 

세 번째 차이는 aiocache 는 TTL(Time-To-Live)을 설정하거나, 키 생성 로직을 커스터마이징 할 수 있습니다. 반면 lru_cache 는 캐시 할 수 있는 항목의 최대 수(maxsize)를 설정하여 오래 사용하지 않은 항목을 삭제하는 방식입니다.

 

 

 

마무리

이 글에서는 aiocache를 사용해 슬랙 API 요청 함수를 캐싱하는 방법을 알아봤습니다. 이러한 캐싱은 API 요청을 최적화하고 제한에 걸리지 않게 관리하는 데 매우 유용합니다. 여러분의 프로젝트에서도 유사한 상황이 있다면 꼭 활용해 보세요! 😄

 

 

---

 

 

참고 문서

https://aiocache.aio-libs.org/en/latest/

https://docs.python.org/3/library/functools.html#functools.lru_cache