beautifulsoups.dev
Published on

Python async 데코레이터로 컨텍스트 깨짐 해결

Authors

서버에서 asyncio 기반으로 API를 처리하다 보면, 분명 같은 요청 흐름인데도 로그의 request_id 가 갑자기 None 이 되거나, 트레이싱 스팬이 끊기고, 사용자 정보가 다른 요청 값으로 섞이는 현상을 종종 만납니다. 현업에서는 이를 보통 “컨텍스트 깨짐”이라고 부릅니다.

이 글에서는 Python async 코드에서 컨텍스트가 왜 깨지는지, 그리고 ContextVar 와 async 데코레이터 조합으로 컨텍스트를 안전하게 전파하고, 필요하면 복구까지 하는 실전 패턴을 정리합니다. FastAPI, Starlette, aiohttp 같은 비동기 프레임워크를 쓰는 경우에도 그대로 적용할 수 있습니다.

아래 내용은 로깅과 관측성(트레이싱)에서 특히 체감이 큽니다. 장애 분석 관점은 AWS IAM AccessDenied 스택추적과 정책 최소화 같은 글에서 다루는 “스택과 맥락을 끝까지 남기는 습관”과도 결이 같습니다.

async에서 컨텍스트가 깨지는 대표 원인

1) 전역 변수로 요청 단위 상태를 들고 있는 경우

가장 흔한 실수는 다음처럼 전역 변수(또는 싱글톤)에 request_id 를 넣는 방식입니다.

# 나쁜 예시
request_id = None

async def handler():
    global request_id
    request_id = "req-123"
    await asyncio.sleep(0)
    print(request_id)

동시 요청이 들어오면 request_id 는 서로 덮어쓰기 때문에 섞입니다.

2) asyncio.create_task 로 분기하면서 컨텍스트를 의도치 않게 분리

Python 3.7+ 에서는 ContextVar 가 Task 생성 시점에 복사되지만, 다음과 같은 패턴에서 “의도한 컨텍스트”가 아닌 값이 들어갈 수 있습니다.

  • create_task 를 호출하는 시점이 늦어져 이미 컨텍스트가 변경됨
  • 라이브러리 내부에서 스레드풀로 넘기며 컨텍스트가 소실
  • 콜백 기반 API에서 컨텍스트를 명시적으로 전달하지 않음

3) 스레드풀(run_in_executor, to_thread)로 넘어가며 끊김

스레드로 넘어가는 순간 ContextVar 가 자동 전파되지 않는 경우가 있습니다(버전, 실행 방식, 프레임워크에 따라 체감이 다릅니다). 특히 CPU 작업이나 블로킹 I/O 를 to_thread 로 처리할 때 “로그 컨텍스트가 갑자기 비어 있음”을 자주 봅니다.

해결의 핵심: ContextVar 를 표준으로 쓰고, 데코레이터로 강제하기

Python에서 요청 단위 컨텍스트를 다루는 가장 안전한 기본은 contextvars.ContextVar 입니다.

  • 전역 변수처럼 보이지만, 실제로는 Task 로컬에 가깝게 동작
  • 비동기 호출 체인에서 안전하게 값이 유지됨
  • 토큰 기반으로 set/reset 이 가능해서 “복구”가 쉬움

먼저 컨텍스트 변수를 정의합니다.

from contextvars import ContextVar

request_id_var: ContextVar[str | None] = ContextVar("request_id", default=None)
user_id_var: ContextVar[str | None] = ContextVar("user_id", default=None)

이제 문제는 팀 코드 전체가 이를 일관되게 쓰도록 강제하는 것입니다. 여기서 async 데코레이터 가 매우 유용합니다.

패턴 1: 컨텍스트를 설정하고 반드시 복구하는 async 데코레이터

요청 핸들러나 유스케이스 함수 진입 시 컨텍스트를 세팅하고, 종료 시 원래 값으로 복구하는 패턴입니다.

from __future__ import annotations

from functools import wraps
from typing import Any, Awaitable, Callable, ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")


def with_request_context(
    *,
    request_id: str | None = None,
    user_id: str | None = None,
) -> Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]:
    def decorator(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
        @wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            token_req = None
            token_user = None

            if request_id is not None:
                token_req = request_id_var.set(request_id)
            if user_id is not None:
                token_user = user_id_var.set(user_id)

            try:
                return await func(*args, **kwargs)
            finally:
                # 반드시 원복해서 상위 컨텍스트를 오염시키지 않음
                if token_user is not None:
                    user_id_var.reset(token_user)
                if token_req is not None:
                    request_id_var.reset(token_req)

        return wrapper

    return decorator

사용 예시는 다음과 같습니다.

@with_request_context(request_id="req-123", user_id="u-9")
async def do_work() -> None:
    await asyncio.sleep(0)
    print("request_id=", request_id_var.get())
    print("user_id=", user_id_var.get())

이 패턴의 장점은 “컨텍스트가 깨지는 것”을 넘어서 상위 호출자의 컨텍스트를 보호한다는 점입니다. 즉, 내부 함수가 컨텍스트를 바꿔도 호출이 끝나면 원래대로 돌아갑니다.

패턴 2: 현재 컨텍스트를 캡처해 Task 생성 시점에 고정

컨텍스트 깨짐은 “Task 생성 시점”과 “Task 실행 시점”이 달라질 때 더 자주 드러납니다. 이를 방지하려면 copy_context 로 현재 컨텍스트를 캡처한 뒤, 그 컨텍스트에서 코루틴을 실행하도록 감싸면 됩니다.

import asyncio
from contextvars import copy_context
from typing import Awaitable, TypeVar

T = TypeVar("T")


def create_task_with_context(coro: Awaitable[T]) -> asyncio.Task[T]:
    ctx = copy_context()

    async def runner() -> T:
        return await coro

    # 컨텍스트를 고정한 상태로 Task 실행
    return asyncio.create_task(ctx.run(runner))

이제 다음처럼 “분기 작업”을 만들어도 컨텍스트가 안정적으로 유지됩니다.

async def handler() -> None:
    request_id_var.set("req-777")

    async def background_job():
        await asyncio.sleep(0)
        print("bg request_id=", request_id_var.get())

    task = create_task_with_context(background_job())
    await task

현업에서는 이 함수를 직접 쓰기보다, 아래처럼 데코레이터로 감싼 “안전한 백그라운드 실행기”를 만들어 공통 유틸로 배포하는 편이 운영에 유리합니다.

패턴 3: async 데코레이터로 로깅 컨텍스트를 자동 주입

대부분의 팀은 로거에 extra 를 붙이거나, 구조화 로깅(JSON) 필드를 채우는 방식으로 컨텍스트를 기록합니다. 문제는 매 함수마다 request_id_var.get() 를 호출해 넣는 게 번거롭고 누락되기 쉽다는 점입니다.

아래는 함수 시작과 종료 시점에 컨텍스트를 자동으로 로그에 남기는 데코레이터 예시입니다.

import logging
from functools import wraps
from typing import Awaitable, Callable, ParamSpec, TypeVar

logger = logging.getLogger(__name__)

P = ParamSpec("P")
R = TypeVar("R")


def log_with_context(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
    @wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        rid = request_id_var.get()
        uid = user_id_var.get()
        logger.info("start %s", func.__name__, extra={"request_id": rid, "user_id": uid})
        try:
            result = await func(*args, **kwargs)
            logger.info("end %s", func.__name__, extra={"request_id": rid, "user_id": uid})
            return result
        except Exception:
            logger.exception("error %s", func.__name__, extra={"request_id": rid, "user_id": uid})
            raise

    return wrapper

이 데코레이터는 “컨텍스트를 깨지 않게 한다”기보다, 깨졌을 때 즉시 관측 가능하게 만드는 효과가 큽니다. 장애 대응에서 로그의 일관성은 체감이 매우 크고, 문제를 끝까지 추적하는 습관은 systemd 서비스가 계속 재시작될 때 원인 9가지 같은 운영 디버깅 글에서 강조하는 관점과도 연결됩니다.

패턴 4: 스레드로 넘어갈 때 컨텍스트를 명시적으로 전파

블로킹 작업을 스레드로 넘길 때 컨텍스트가 끊기면, “스레드에서 찍힌 로그만 request_id가 없다” 같은 상황이 생깁니다. 이때도 copy_context 로 해결할 수 있습니다.

import asyncio
from contextvars import copy_context
from typing import Callable, TypeVar

T = TypeVar("T")


def to_thread_with_context(func: Callable[[], T]) -> "asyncio.Future[T]":
    ctx = copy_context()
    return asyncio.to_thread(ctx.run, func)

사용 예시:

def blocking_io() -> str:
    # 여기서도 request_id_var.get() 이 유지되길 기대
    return f"blocking rid={request_id_var.get()}"

async def handler() -> None:
    request_id_var.set("req-999")
    result = await to_thread_with_context(blocking_io)
    print(result)

이 패턴은 특히 “레거시 라이브러리 호출” 또는 “SDK 내부가 동기 함수인 경우”에 유용합니다.

흔한 실수와 체크리스트

데코레이터에서 await 를 빼먹는 실수

비동기 함수를 감싸면서 return func(*args, **kwargs) 로 반환해버리면, 호출자는 코루틴을 받게 되고 실행 시점이 바뀌어 컨텍스트가 어긋날 수 있습니다. 반드시 await 하세요.

ContextVar.set 후 reset 을 안 하는 실수

특히 “라이브러리 함수”에서 컨텍스트를 설정해두고 원복하지 않으면, 같은 워커에서 다음 요청이 이전 요청 컨텍스트를 물고 시작하는 형태의 오염이 발생할 수 있습니다. 토큰을 받아 finally 에서 reset 하는 구조를 습관화하세요.

컨텍스트에 큰 객체를 넣는 실수

request 전체 객체, 대형 payload, DB 세션 같은 것을 컨텍스트에 넣으면 메모리 사용량과 참조 유지 문제가 커집니다. 컨텍스트에는 보통 request_id, user_id, trace_id 처럼 작은 식별자만 넣고, 큰 객체는 명시적으로 전달하는 편이 안전합니다.

FastAPI 스타일로 적용하는 실전 구성 예시

FastAPI에서는 미들웨어에서 request_id 를 만들고, 핸들러 체인에서 자동으로 유지되길 기대합니다. 아래는 개념 예시입니다.

import uuid
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    rid = request.headers.get("x-request-id") or str(uuid.uuid4())
    token = request_id_var.set(rid)
    try:
        response = await call_next(request)
        response.headers["x-request-id"] = rid
        return response
    finally:
        request_id_var.reset(token)

그리고 서비스 함수에는 @log_with_context 같은 데코레이터를 붙여 “컨텍스트 누락”을 빠르게 드러내게 만들 수 있습니다.

@app.get("/items")
@log_with_context
async def list_items():
    # 어디서든 request_id_var.get() 가능
    return {"request_id": request_id_var.get()}

마무리: async 컨텍스트는 “자동”이 아니라 “설계”다

Python 비동기 환경에서 컨텍스트는 생각보다 쉽게 깨집니다. 특히 작업 분기(create_task), 스레드 전환(to_thread), 라이브러리 내부 실행 모델 차이에서 문제가 자주 발생합니다.

정리하면 다음 3가지를 표준으로 두면 운영 난이도가 확 내려갑니다.

  1. 요청 단위 상태는 전역 변수가 아니라 ContextVar 로 관리
  2. 컨텍스트 설정은 데코레이터나 미들웨어로 공통화하고, reset 으로 반드시 복구
  3. Task 생성과 스레드 전환 지점에는 copy_context 기반 유틸을 제공

비슷한 결로 “문제가 생겼을 때 맥락을 남기고 재현 가능하게 만드는 방법”은 장애 대응에 직결됩니다. 재시도나 폴백 설계 관점은 Claude API 529 Overloaded 재시도·폴백 설계도 함께 참고하면 좋습니다.

Discuss on Twitter • View on GitHub