Python async context manager 3종 패턴과 함정

비동기 I/O가 기본인 환경에서는 try/finally 를 매번 손으로 쓰기보다 async with 로 자원 수명(lifetime)을 강제하는 편이 안전합니다. 문제는 async context manager를 “어떻게 구현하느냐”에 따라 예외 전파, 취소(CancelledError) 처리, 세마포어/락 해제 타이밍, 커넥션 반환 같은 핵심 동작이 미묘하게 달라진다는 점입니다.

이 글에서는 Python에서 자주 쓰는 async context manager 3종 패턴을 정리하고, 실무에서 자주 밟는 함정을 코드로 짚습니다. (특히 타임아웃/취소/예외 억제 관련)

참고로 비동기에서 자원 누수는 언어를 가리지 않습니다. Go에서도 채널 블로킹으로 고루틴이 누수되는 케이스가 흔하니, 비슷한 관점으로 함께 보면 좋습니다: Go 고루틴 누수 원인 - 채널 블로킹 5가지

async context manager가 보장해야 하는 것

async with mgr as x: 는 내부적으로 대략 다음 순서로 동작합니다.

1. await mgr.__aenter__() 실행 후 반환값을 x 에 바인딩
2. 본문 실행
3. 예외가 나든 말든 await mgr.__aexit__(exc_type, exc, tb) 호출
4. __aexit__ 가 True 를 반환하면 예외를 “억제(suppress)” 하고, False 면 예외를 전파

핵심 체크리스트는 다음입니다.

• 실패해도 정리(cleanup)가 실행되는가
• 정리 중 예외가 원래 예외를 덮어쓰지 않는가
• 취소(asyncio.CancelledError)가 들어왔을 때 락/세마포어가 확실히 풀리는가
• __aexit__ 에서 예외를 실수로 삼키지 않는가

패턴 1: 클래스 기반 __aenter__/__aexit__ (정석이자 가장 명시적)

가장 전통적인 방식입니다. 상태를 객체에 보관하기 쉽고, 타입 힌트/테스트도 안정적입니다.

예제: 비동기 락을 안전하게 감싸기

import asyncio
from typing import Optional, Type

class AsyncLockGuard:
    def __init__(self, lock: asyncio.Lock):
        self._lock = lock
        self._acquired = False

    async def __aenter__(self) -> "AsyncLockGuard":
        await self._lock.acquire()
        self._acquired = True
        return self

    async def __aexit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        tb,
    ) -> bool:
        if self._acquired:
            self._lock.release()
            self._acquired = False
        # 예외를 억제하지 않고 그대로 전파
        return False

async def main():
    lock = asyncio.Lock()

    async with AsyncLockGuard(lock):
        # 임계구역
        await asyncio.sleep(0.1)

asyncio.run(main())

함정 1) __aexit__ 에서 예외를 무심코 억제하기

__aexit__ 가 True 를 반환하면 예외가 사라집니다. 디버깅이 매우 어려워지는 전형적인 실수입니다.

class BadGuard:
    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc, tb):
        # 실수: 항상 True를 반환하면 모든 예외가 사라짐
        return True

권장: 특별히 의도한 경우(예: 특정 예외만 무시) 외에는 return False 를 기본으로 두세요.

함정 2) 정리(cleanup) 중 예외가 원래 예외를 덮어쓰기

본문에서 발생한 예외가 더 중요한데, __aexit__ 에서 또 예외가 나면 “정리 예외”가 원래 예외를 가려버립니다.

해결책은 두 가지입니다.

• 정리 단계는 최대한 예외가 나지 않게 설계
• 불가피하다면 정리 예외를 로깅하고 원래 예외를 유지

import logging

logger = logging.getLogger(__name__)

class SaferCleanup:
    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc, tb):
        try:
            await self._cleanup()
        except Exception:
            logger.exception("cleanup failed")
            # 원래 예외를 억제하지 않음
        return False

    async def _cleanup(self):
        ...

함정 3) 취소(CancelledError)에서 락이 풀리지 않는 경우

대부분의 finally 는 취소에도 실행되지만, “취소를 잡아서 삼키는 코드”가 섞이면 락이 영구적으로 잡히기도 합니다.

원칙:

• CancelledError 는 특별한 이유가 없으면 다시 raise
• __aexit__ 에서 return True 로 취소를 억제하지 말 것

이 문제는 재시도/백오프 로직에서도 자주 같이 터집니다. 타임아웃/취소/재시도 설계는 다음 글도 함께 참고할 만합니다: OpenAI 429 rate_limit_exceeded 재시도·백오프 설계

패턴 2: contextlib.asynccontextmanager (간결하지만 제어 흐름 함정이 있음)

contextlib.asynccontextmanager 데코레이터는 yield 를 경계로 진입/정리 코드를 나눠 쓰는 방식입니다. 짧고 읽기 쉬워서 실무에서 가장 많이 보입니다.

예제: 연결을 열고 닫는 패턴

from contextlib import asynccontextmanager

class FakeConn:
    async def open(self):
        ...
    async def close(self):
        ...

@asynccontextmanager
async def open_conn():
    conn = FakeConn()
    await conn.open()
    try:
        yield conn
    finally:
        await conn.close()

async def handler():
    async with open_conn() as conn:
        ...

함정 1) yield 를 두 번 하거나, yield 뒤에 도달하지 못하는 흐름

asynccontextmanager 는 “정확히 한 번 yield” 해야 합니다.

• yield 를 두 번 하면 런타임 에러
• yield 이전에 예외가 나면 진입 자체가 실패하며, 외부에서는 __aenter__ 실패로 보임

또한 yield 이후 코드는 무조건 “정리 단계”입니다. 정리 단계에서 오래 걸리는 I/O를 하면 호출자 입장에서 async with 블록 종료가 지연됩니다.

함정 2) 예외 억제를 잘못 구현하기

asynccontextmanager 에서 예외를 억제하려면 try/except 로 잡고 “다시 raise하지 않으면” 됩니다. 그런데 이게 너무 쉽기 때문에, 의도치 않게 예외를 삼키기 쉽습니다.

@asynccontextmanager
async def suppress_all_accidentally():
    try:
        yield
    except Exception:
        # 실수: 로깅만 하고 끝내면 예외가 사라짐
        return

권장 패턴은 “무시할 예외만 명시적으로” 처리하는 것입니다.

@asynccontextmanager
async def suppress_only_timeout():
    try:
        yield
    except TimeoutError:
        # 의도적으로 무시
        return

함정 3) cleanup에서 또 다른 예외가 나면 원래 예외가 가려짐

클래스 패턴과 동일합니다. finally 내부에서 예외가 터지면 본문 예외가 덮일 수 있습니다. 정리 코드는 실패 가능성을 최소화하고, 실패해도 로깅 후 원래 예외를 유지하도록 구성하세요.

패턴 3: “획득/해제”를 조합하는 스택 패턴 (AsyncExitStack)

복수의 자원을 조건적으로 열고, 열었으면 역순으로 닫아야 하는 경우가 많습니다. 예를 들어:

• 옵션에 따라 파일/소켓/DB 트랜잭션을 다르게 열기
• 여러 락을 순서대로 획득하고 실패 시 이미 획득한 것만 해제
• 중간 단계에서 예외가 나도 이미 열린 것들은 모두 정리

이때 contextlib.AsyncExitStack 이 가장 안전합니다.

예제: 조건부로 여러 자원 관리

from contextlib import AsyncExitStack, asynccontextmanager

class Resource:
    def __init__(self, name: str):
        self.name = name
    async def open(self):
        ...
    async def close(self):
        ...

@asynccontextmanager
async def managed_resource(name: str):
    r = Resource(name)
    await r.open()
    try:
        yield r
    finally:
        await r.close()

async def do_work(use_extra: bool):
    async with AsyncExitStack() as stack:
        primary = await stack.enter_async_context(managed_resource("primary"))

        extra = None
        if use_extra:
            extra = await stack.enter_async_context(managed_resource("extra"))

        # 여기서 예외가 나도 primary/extra는 역순으로 close
        ...

함정 1) enter_async_context 를 await 하지 않기

enter_async_context 는 코루틴을 반환하므로 반드시 await 해야 합니다. 실수로 빼먹으면 자원이 열리지도 않았는데 변수에 코루틴 객체가 들어가고, 정리는 더 혼란스러워집니다.

# 잘못된 예
primary = stack.enter_async_context(managed_resource("primary"))  # await 누락

함정 2) 정리 순서(역순)를 고려하지 않은 설계

AsyncExitStack 은 “들어간 역순”으로 정리합니다. 의존성이 있는 자원(예: 트랜잭션이 커넥션에 의존)이라면 들어가는 순서를 의도적으로 설계해야 합니다.

• 커넥션 open 후 트랜잭션 begin
• 정리는 트랜잭션 rollback/commit 후 커넥션 close

함정 3) 부분 성공/부분 실패 시 보상 로직을 섞을 때

여러 자원을 열고 작업을 진행하다 중간 실패하면 “이미 수행한 작업을 되돌리는 보상 트랜잭션”이 필요해질 수 있습니다. 이때 context manager의 정리는 “자원 해제”에 집중하고, “업무 보상”은 별도 계층에서 명확히 다루는 편이 좋습니다.

보상 트랜잭션 중복 실행 방지는 다음 글의 아이디어를 비동기 작업에도 그대로 적용할 수 있습니다: Saga 패턴 보상 트랜잭션 중복 실행 방지법

실무에서 자주 겪는 공통 함정 6가지

1) async with 안에서 또 다른 태스크를 만들고 정리를 기대하기

async with 가 보장하는 정리는 “현재 태스크의 제어 흐름”에 한정됩니다. 블록 안에서 asyncio.create_task(...) 로 백그라운드 작업을 만들고, 그 작업이 같은 자원을 쓰면:

• 블록이 끝나며 자원이 닫힌 뒤에도 백그라운드 태스크가 접근
• 간헐적으로 closed 류 에러, 데이터 손상, 레이스 컨디션

해결: 백그라운드 태스크가 자원을 쓴다면 태스크의 생명주기도 함께 묶어 관리하세요(태스크 그룹, 명시적 cancel/join 등).

2) 타임아웃과 정리의 순서 문제

타임아웃으로 본문이 끊길 때 정리가 지연되거나, 정리 자체가 타임아웃에 걸려 더 큰 문제를 만들 수 있습니다. 정리 단계는 가능하면 “짧고 실패하지 않게” 설계하고, 정리 타임아웃을 별도로 두는 것도 고려하세요.

3) 예외 타입에 따라 다른 정리 정책이 필요한데 한 덩어리로 처리

예:

• 정상 종료면 commit
• 예외면 rollback
• 취소면 rollback 하되 취소는 다시 raise

이런 분기는 __aexit__ 의 exc_type 을 기반으로 명확히 나눠야 합니다.

import asyncio

class Tx:
    async def begin(self): ...
    async def commit(self): ...
    async def rollback(self): ...

class TxContext:
    def __init__(self, tx: Tx):
        self.tx = tx

    async def __aenter__(self):
        await self.tx.begin()
        return self.tx

    async def __aexit__(self, exc_type, exc, tb):
        if exc_type is None:
            await self.tx.commit()
            return False

        # 취소는 롤백 후 반드시 전파
        if exc_type is asyncio.CancelledError:
            await self.tx.rollback()
            return False

        await self.tx.rollback()
        return False

4) “정리 실패”를 무시해도 되는지 정책이 없음

정리 실패는 운영 이슈로 이어질 수 있습니다.

• 커넥션 반환 실패 -> 풀 고갈
• 락 해제 실패 -> 데드락

정리 실패를 로깅만 하고 넘어갈지, 프로세스를 죽일지, 재시도할지 기준을 정하세요.

주의: 본문에 -> 같은 기호를 그대로 쓰면 MDX에서 문제가 될 수 있으니, 위처럼 인라인 코드로 감싸는 습관이 안전합니다.

5) async generator와 async context manager를 혼동

asynccontextmanager 는 내부적으로 async generator를 쓰지만, “그냥 async generator”를 async with 에 넣을 수는 없습니다. 반드시 데코레이터로 감싸 context manager 프로토콜을 제공해야 합니다.

6) 테스트에서만 통과하고 운영에서만 누수되는 케이스

테스트는 보통 단일 스레드/단일 루프/낮은 동시성이라 타이밍 버그가 숨습니다. 다음을 체크하면 누수/경합을 빨리 잡을 수 있습니다.

• 동시성(예: 100~1000 태스크)으로 부하 테스트
• 타임아웃/취소를 인위적으로 주입
• 풀/세마포어/락의 현재 값(대기자 수 포함) 메트릭화

어떤 패턴을 언제 쓰면 좋은가

• 클래스 기반: 상태가 많고, 타입/정책(예외 분기, 메트릭, 로깅)을 명시적으로 두고 싶을 때. 라이브러리 품질로 제공할 때 가장 무난.
• asynccontextmanager: 단일 자원에 대해 “열기-사용-닫기”가 간단할 때. 단, 예외 억제 실수와 cleanup 예외 덮어쓰기에 주의.
• AsyncExitStack: 조건부/다중 자원 조합, 부분 실패 롤백, 역순 정리가 필요한 복잡한 플로우에서 최강.

마무리

async context manager는 단순히 문법 설탕이 아니라, 비동기 시스템에서 “자원 수명과 예외 전파 정책”을 코드로 고정하는 장치입니다. 구현 패턴을 선택할 때는 코드 길이보다도 다음을 우선하세요.

• 예외를 숨기지 않는가
• 취소에도 자원이 확실히 정리되는가
• 정리 단계가 원래 예외를 덮지 않는가
• 다중 자원에서 부분 실패를 안전하게 처리하는가

위 3가지 패턴을 상황에 맞게 섞어 쓰면, 비동기 코드에서 가장 고통스러운 유형의 버그(간헐적 누수/교착/풀 고갈)를 크게 줄일 수 있습니다.