[Python] contextvars

fastapi가 등장한 이후로, python에서도 비동기로 구현하는 방식이 많이 활성화되고 있습니다.

멀티 스레딩으로 구현할 때는, thread의 context를 활용해서 로그를 남겨서 쉽게 모니터링 할 수 있었는데 비동기에서는 thread context가 없어서 해당 부분을 해소하기 위해서 등장한 것이 contextvars입니다.

 

contextvars는 파이썬 표준 라이브러리로 따로 설치를 해야하는 것은 없습니다.

사용법은 매우 간단합니다.

ContextVar를 한 번 선언을 하고, set 함수를 호출하여 context를 설정해주면 get 함수를 호출해서 가져다 쓸 수 있습니다.

 

바로 예시 코드를 보겠습니다.

import asyncio
import contextvars
import random
import uuid

context = contextvars.ContextVar("context", default="context")


async def job(job_id: int):
  context.set(str(uuid.uuid4()))
  print(f"1 - job id: {job_id}, context: {context.get()}")
  await asyncio.sleep(random.randint(1, 5))
  print(f"2 - job id: {job_id}, context: {context.get()}")


async def main():
  await asyncio.gather(*[job(i) for i in range(5)])
  
  
if __name__ == "__main__":
  asyncio.run(main())

job이라는 함수를 5번 동시에 실행시키는 코드입니다.

job이라는 함수에서는 context를 설정해주고, 한번 출력 후 random sleep 이후에 다시 context를 출력하게 됩니다.

실행해서 보시면, context가 잘 유지되어서 출력되는 것을 확인할 수 있습니다.

 

fastapi에서는 async로 구현하게 되는데, api가 호출되고 부터 함수를 넘나드는 동안 로그를 찍다보면 동일한 api에서의 호출인 지 확인할 수 있는 방법이 따로 없습니다.

contextvars를 응용해서, fastapi의 Middleware에서 처음에 context를 설정하게 구현하고 로그를 찍을 때 context를 항상 같이 출력하게 되면, api가 호출되어 응답이 나갈 때 까지 동일한 context로 출력이 되기 때문에, context를 기준으로 로그를 검색하면 api 한 번 호출에 대한 모든 로그를 한 눈에 볼 수 있게 됩니다.