beautifulsoups.dev
Published on

Python async Context Manager 3패턴·함정

Authors

서버 사이드 Python에서 async with는 단순히 문법 설탕이 아니라, 비동기 자원 수명주기(lifecycle)와 예외 안전성을 보장하는 핵심 도구입니다. DB 커넥션, HTTP 세션, 분산락, 임시 파일, 트레이싱 스팬 같은 것들은 대부분 “열고 닫기”가 필요하고, 그 사이에 await가 끼어드는 순간 동기 with와는 다른 함정이 생깁니다.

이 글에서는 실무에서 가장 많이 쓰는 3가지 async context manager 패턴과, 운영에서 사고로 이어지기 쉬운 함정들을 정리합니다.

또한 비동기 코드의 문제는 재현이 어렵고 로그도 산발적인 경우가 많으니, 운영 트러블슈팅 관점에서는 다른 런타임의 디버깅 글도 함께 참고하면 좋습니다. 예를 들어 동시성 문제를 빠르게 좁히는 접근은 Go 채널 데드락 10분 디버깅 - select+context에서 아이디어를 얻을 수 있습니다.

async context manager 기본: 무엇이 호출되나

async with obj:는 내부적으로 다음을 호출합니다.

  • await obj.__aenter__()
  • 블록 실행
  • await obj.__aexit__(exc_type, exc, tb)

즉 enter/exit 모두 await 가능하며, 블록 내부에서 예외가 나거나 태스크가 취소되어도 __aexit__가 호출됩니다(단, 아래에서 설명할 “취소 전파” 함정이 있음).

간단한 형태는 다음과 같습니다.

class MyAsyncCM:
    async def __aenter__(self):
        print("enter")
        return self

    async def __aexit__(self, exc_type, exc, tb):
        print("exit", exc_type)
        return False  # 예외를 삼키지 않음


async def main():
    async with MyAsyncCM():
        raise RuntimeError("boom")

__aexit__의 반환값이 True면 예외를 억제(suppress)하고, False면 예외를 그대로 전파합니다.

패턴 1) 클래스로 구현: 자원 보유 객체에 가장 명확한 방식

언제 쓰나

  • 커넥션, 락, 세마포어처럼 “자원 자체가 객체의 정체성”인 경우
  • enter에서 자원을 획득하고, exit에서 해제하는 수명주기가 명확한 경우
  • 테스트에서 목 객체를 주입하고 싶을 때

예시: 분산락을 흉내낸 async context manager

import asyncio

class AsyncLock:
    def __init__(self, lock: asyncio.Lock):
        self._lock = lock

    async def __aenter__(self):
        await self._lock.acquire()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        self._lock.release()
        return False


async def worker(lock: asyncio.Lock):
    async with AsyncLock(lock):
        await asyncio.sleep(0.1)

함정: __aenter__에서 부분 성공 후 실패

__aenter__에서 여러 자원을 단계적으로 열다가 중간에 예외가 나면, 이미 확보한 자원이 누수될 수 있습니다. 예를 들어 “소켓 연결 성공 후 인증 실패” 같은 케이스가 대표적입니다.

대응은 보통 둘 중 하나입니다.

  • __aenter__ 내부에서 try/finally로 부분 성공 자원을 정리
  • 더 안전하게, “여러 자원”은 AsyncExitStack으로 관리(패턴 3에서 설명)

패턴 2) @asynccontextmanager: 짧고 읽기 쉬운 제너레이터 기반

표준 라이브러리 contextlib.asynccontextmanager는 한 번 yield하는 async generator로 컨텍스트 매니저를 만들 수 있게 해줍니다.

언제 쓰나

  • 함수 단위로 자원 스코프를 간단히 만들고 싶을 때
  • “열기/닫기” 로직이 짧고 명확할 때
  • 테스트에서 쉽게 대체하고 싶을 때

예시: aiohttp 세션 스코프

from contextlib import asynccontextmanager
import aiohttp

@asynccontextmanager
async def http_session():
    session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10))
    try:
        yield session
    finally:
        await session.close()


async def fetch_json(url: str):
    async with http_session() as s:
        async with s.get(url) as resp:
            resp.raise_for_status()
            return await resp.json()

함정 1: yield 이전에 예외가 나면 finally가 실행되지 않음

@asynccontextmanager는 “yield 이후”에 finally가 보장됩니다. 즉, 아래처럼 yield 전에 예외가 나면 정리 코드가 실행되지 않습니다.

from contextlib import asynccontextmanager

@asynccontextmanager
async def broken():
    resource = await open_resource()
    await do_something_that_may_fail()  # 여기서 예외가 나면
    try:
        yield resource
    finally:
        await resource.close()  # 실행되지 않을 수 있음

해결은 yield 이전 단계도 try/finally에 포함시키는 것입니다.

from contextlib import asynccontextmanager

@asynccontextmanager
async def safe_cm():
    resource = await open_resource()
    try:
        await do_something_that_may_fail()
        yield resource
    finally:
        await resource.close()

함정 2: 예외 억제 로직을 실수로 만들어버림

클래스 방식에서는 __aexit__의 반환값으로 억제를 명시하지만, @asynccontextmanager에서는 try/except를 잘못 쓰면 예외를 조용히 삼킬 수 있습니다.

@asynccontextmanager
async def swallow_errors():
    try:
        yield
    except Exception:
        # 로깅만 하고 끝내면 호출자는 성공으로 오해
        return

실무에서는 예외를 억제할 이유가 거의 없고, 억제하더라도 특정 예외만 엄격히 제한하는 것이 안전합니다.

패턴 3) AsyncExitStack: 여러 자원을 조합하는 “트랜잭션형 스코프”

여러 자원을 한 스코프에서 다루면 문제가 급격히 어려워집니다.

  • 1번 자원은 성공
  • 2번 자원은 성공
  • 3번 자원 획득 중 실패

이때 1, 2번을 역순으로 정리해야 합니다. 동기 세계에서는 ExitStack이, 비동기에서는 AsyncExitStack이 이 문제를 해결합니다.

언제 쓰나

  • “N개 자원을 조건부로 열고, 실패 시 열린 것만 닫기”가 필요할 때
  • 플러그인 구조처럼 런타임에 어떤 자원을 열지 동적으로 결정할 때
  • DB 트랜잭션, 메시지 ack, 임시 파일 삭제 등을 하나의 스코프로 묶고 싶을 때

예시: 조건부로 여러 리소스 묶기

from contextlib import AsyncExitStack, asynccontextmanager

@asynccontextmanager
async def open_a(name: str):
    a = await open_resource_a(name)
    try:
        yield a
    finally:
        await a.close()

@asynccontextmanager
async def open_b(flag: bool):
    b = await open_resource_b() if flag else None
    try:
        yield b
    finally:
        if b is not None:
            await b.close()


async def use_many(flag: bool):
    async with AsyncExitStack() as stack:
        a = await stack.enter_async_context(open_a("primary"))
        b = await stack.enter_async_context(open_b(flag))

        # 여기서 어떤 예외가 나도 a, b는 역순으로 정리됨
        await a.do()
        if b:
            await b.do()

함정: “부분 성공” 정리가 누락되는 구조적 실수 방지

AsyncExitStack의 핵심 장점은, 중간 단계에서 예외가 나도 그 시점까지 stack에 등록된 exit 콜백이 모두 실행된다는 점입니다. 즉, 패턴 1과 2에서 언급한 “enter 중 실패” 누수를 구조적으로 줄여줍니다.

운영에서 이런 누수는 결국 커넥션 고갈이나 파일 디스크립터 고갈로 이어져 장애가 되고, 증상은 “갑자기 재시작 반복” 같은 형태로 나타나기도 합니다. 이런 류의 장애 징후를 보는 관점은 systemd 서비스가 계속 재시작될 때 진단 체크리스트도 함께 보면 도움이 됩니다.

실무 함정 모음: 취소, 타임아웃, 백그라운드 태스크

함정 A) 취소가 __aexit__를 방해할 수 있다

asyncio에서 태스크 취소는 보통 CancelledError로 전달됩니다. 문제는 __aexit__ 안에서 await를 하는 동안에도 취소가 다시 들어올 수 있다는 점입니다. 그러면 “정리 로직이 끝나기 전에 또 취소”가 걸려 자원 해제가 중단될 수 있습니다.

대응 방법 중 하나는 정리 구간을 asyncio.shield로 감싸 정리 작업만큼은 취소 전파를 막는 것입니다.

import asyncio

class SafeCloser:
    def __init__(self, res):
        self.res = res

    async def __aenter__(self):
        return self.res

    async def __aexit__(self, exc_type, exc, tb):
        # close가 반드시 끝나야 하는 경우
        await asyncio.shield(self.res.aclose())
        return False

주의할 점은, shield는 “취소를 무시”하는 게 아니라 “바깥 취소가 안쪽 await에 바로 전파되지 않게” 하는 것입니다. 정리 시간이 너무 길면 shutdown이 지연될 수 있으므로, close 자체에 타임아웃을 두는 것도 고려하세요.

함정 B) 타임아웃 컨텍스트와 중첩 순서

asyncio.timeout() 같은 타임아웃 컨텍스트와 자원 컨텍스트를 중첩할 때 순서가 중요합니다. 타임아웃이 먼저 걸려 취소가 발생하면, 내부 자원 정리까지 취소가 전파될 수 있습니다.

권장 형태는 “타임아웃은 작업 범위를 감싸되, 정리는 보호”입니다.

import asyncio

async def do_work(res):
    async with res:  # res가 async context manager라고 가정
        async with asyncio.timeout(2):
            await res.query()

정리 자체가 오래 걸릴 수 있다면, 앞서 말한 shield 또는 close 타임아웃을 조합하세요.

함정 C) 컨텍스트 밖으로 새 태스크를 만들어 누수시키기

async with 블록 안에서 asyncio.create_task()로 백그라운드 작업을 만들고, 그 작업이 컨텍스트가 닫힌 뒤에도 자원을 쓰면 레이스 컨디션이 됩니다.

async def handler(pool):
    async with pool.acquire() as conn:
        asyncio.create_task(conn.execute("SELECT 1"))
        # 블록을 나가면 conn이 반환될 수 있음

해결책은 다음 중 하나입니다.

  • 백그라운드 태스크를 만들지 말고 블록 안에서 await로 끝내기
  • 꼭 필요하면 태스크를 추적하고, 블록을 나가기 전에 await task 또는 취소 후 정리
  • 구조화된 동시성(예: TaskGroup)을 사용해 스코프를 강제
import asyncio

async def handler(pool):
    async with pool.acquire() as conn:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(conn.execute("SELECT 1"))
            tg.create_task(conn.execute("SELECT 2"))
        # TaskGroup이 join된 뒤에야 블록을 나감

체크리스트: 안전한 async with를 위한 7가지 규칙

  1. __aenter__에서 여러 단계를 수행한다면 “부분 성공 정리”를 고려한다.
  2. 단순한 스코프는 @asynccontextmanager로 만들되, yield 이전 예외에 주의한다.
  3. 여러 자원을 조합하면 AsyncExitStack으로 구조적으로 누수를 막는다.
  4. __aexit__에서의 await는 취소에 취약하다. 정말 중요한 정리는 asyncio.shield를 검토한다.
  5. 타임아웃 컨텍스트와 자원 컨텍스트의 중첩 순서를 의식한다.
  6. 컨텍스트 내부 자원을 컨텍스트 밖으로 “탈출”시키는 백그라운드 태스크를 경계한다.
  7. 예외 억제는 기본적으로 하지 말고, 필요하면 예외 타입을 좁혀 명시적으로 처리한다.

마무리

Python의 async context manager는 “자원을 안전하게 닫는 문법”을 넘어, 동시성 환경에서의 일관된 종료와 실패 처리를 강제하는 장치입니다. 클래스 구현, @asynccontextmanager, AsyncExitStack 3가지를 상황에 맞게 고르면 코드가 단단해지고, 특히 장애 상황에서 누수와 레이스를 크게 줄일 수 있습니다.

비동기 자원 누수는 결국 메모리, FD, 커넥션 풀 고갈로 이어져 시스템 전체 증상을 만들 수 있습니다. 비슷한 관점의 누수 추적 접근이 궁금하다면 Go gRPC 메모리 누수? pprof로 잡는 7단계도 함께 읽어보면, “관측 가능성부터 가설 검증까지”의 흐름을 잡는 데 도움이 됩니다.

Discuss on Twitter • View on GitHub