beautifulsoups.dev
Published on

contextlib로 async Context Manager 3줄 패턴

Authors

서버 개발에서 비동기 리소스 관리는 늘 반복됩니다. 예를 들어 aiohttp.ClientSession을 열고 닫기, DB 세션 트랜잭션 커밋·롤백, 분산 락 획득·해제, 임시 파일 생성·삭제 같은 작업은 결국 다음의 공통 구조로 수렴합니다.

  • 준비 단계에서 리소스를 만들고
  • 본문에서 사용한 뒤
  • 성공이면 정상 종료, 실패면 정리 로직을 수행

이걸 매번 try/finally로 풀어 쓰면 코드가 길어지고, 예외 분기마다 정리 로직이 누락되기 쉽습니다. Python의 contextlib.asynccontextmanager는 이 반복을 “3줄 패턴”으로 압축해 주는 도구입니다.

아래에서는 이 패턴의 핵심, 예외 처리 방식, 실무에서 바로 쓰는 예제, 그리고 흔한 실수까지 한 번에 정리합니다.

왜 asynccontextmanager가 필요한가

비동기 코드에서는 리소스가 다음과 같은 형태로 존재합니다.

  • 네트워크 커넥션: HTTP 클라이언트, 메시지 브로커
  • DB 커넥션/세션: 커밋·롤백 규칙이 있는 상태 머신
  • 동시성 제어: asyncio.Lock 같은 락
  • 관측/로깅: span 열고 닫기, 타이머 측정

이 리소스들은 “열었으면 반드시 닫아야” 합니다. 특히 예외가 발생해도 정리 단계는 보장돼야 합니다.

전통적인 방식은 다음처럼 try/finally 입니다.

session = aiohttp.ClientSession()
try:
    ...
finally:
    await session.close()

나쁘지 않지만, 리소스 종류가 늘어나면 중첩이 깊어지고, 커밋·롤백 같은 정책이 섞이면서 실수하기 쉬워집니다.

3줄 패턴의 정체

contextlib.asynccontextmanager는 “yield 전후를 진입/종료로 해석”합니다. 즉, yield 이전은 __aenter__, yield 이후는 __aexit__ 역할을 합니다.

가장 작은 형태는 다음입니다.

from contextlib import asynccontextmanager

@asynccontextmanager
async def resource():
    r = await open_resource()
    try:
        yield r
    finally:
        await close_resource(r)

여기서 말하는 “3줄 패턴”은 보통 아래 세 덩어리를 뜻합니다.

  1. r = ... 준비
  2. yield r 제공
  3. finally: ... 정리

실무에서는 준비/정리 내부가 길어질 수 있지만, 구조 자체는 이 3줄로 고정됩니다. 이 구조가 유지되면 리뷰할 때도 “리소스가 반드시 닫히는지”를 빠르게 확인할 수 있습니다.

기본 예제: aiohttp 세션을 3줄로 감싸기

from contextlib import asynccontextmanager
import aiohttp

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

사용부는 깔끔해집니다.

async def fetch_json(url: str) -> dict:
    async with http_session() as session:
        async with session.get(url) as resp:
            resp.raise_for_status()
            return await resp.json()

포인트는 “열고 닫는 책임”을 함수 하나로 캡슐화했다는 점입니다. 호출자는 성공/실패를 신경 쓰지 않고 async with만 쓰면 됩니다.

예외 처리 정책을 넣는 법: 커밋·롤백 패턴

DB 세션은 단순히 닫는 것만으로 끝나지 않습니다. 성공 시 커밋, 예외 시 롤백이라는 정책이 필요합니다.

asynccontextmanager에서는 yield를 기준으로 예외가 전파되므로, try/except/finally를 다음처럼 배치합니다.

from contextlib import asynccontextmanager

@asynccontextmanager
async def unit_of_work(session_factory):
    session = session_factory()
    try:
        yield session
        await session.commit()
    except Exception:
        await session.rollback()
        raise
    finally:
        await session.close()

사용부는 트랜잭션 정책을 몰라도 됩니다.

async def create_user(session_factory, user):
    async with unit_of_work(session_factory) as session:
        session.add(user)

이 패턴을 사용하면 “예외가 나면 반드시 롤백”이 강제됩니다. 분산 트랜잭션이나 사가 패턴에서도 로컬 트랜잭션 경계를 명확히 하는 게 중요합니다. 트랜잭션 경계를 설계할 때는 사가/아웃박스 같은 접근도 함께 참고하면 좋습니다.

동시성 제어: 락을 안전하게 감싸기

asyncio.Lock는 자체적으로 async with lock:을 지원하지만, 실무에서는 “락 획득 전에 로깅/메트릭”, “타임아웃”, “락 이름 표준화” 같은 부가 정책이 들어갑니다. 이때 3줄 패턴이 빛납니다.

from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def acquire(lock: asyncio.Lock, timeout_s: float = 1.0):
    try:
        await asyncio.wait_for(lock.acquire(), timeout=timeout_s)
        yield
    finally:
        if lock.locked():
            lock.release()

사용 예:

async def critical_section(lock: asyncio.Lock):
    async with acquire(lock, timeout_s=0.5):
        # 경쟁 조건이 생기면 안 되는 구역
        ...

주의: finally에서 lock.locked()를 확인하는 이유는 “타임아웃으로 acquire 실패”한 경우 release가 예외를 낼 수 있기 때문입니다.

재시도·백오프와 결합: 실패를 ‘정리 가능한 실패’로 만들기

API 호출이나 외부 시스템 연동은 실패가 잦습니다. 재시도 로직을 호출부마다 흩뿌리면 복잡도가 급격히 늘어납니다. 리소스 관리와 재시도를 같이 설계하면, 실패 시에도 “정리 로직이 항상 실행”되면서 재시도가 가능해집니다.

예를 들어, 특정 작업 단위를 “시도 단위 컨텍스트”로 감싸서 관측 지표를 남길 수 있습니다.

from contextlib import asynccontextmanager
import time

@asynccontextmanager
async def attempt_span(name: str, attempt: int):
    start = time.perf_counter()
    try:
        yield
        ok = True
    except Exception:
        ok = False
        raise
    finally:
        elapsed_ms = (time.perf_counter() - start) * 1000
        # 여기서 로그/메트릭을 남긴다고 가정
        # log.info("span", name=name, attempt=attempt, ok=ok, elapsed_ms=elapsed_ms)
        _ = (name, attempt, ok, elapsed_ms)

재시도 루프에서 사용:

import asyncio

async def call_with_retry(fn, retries: int = 3):
    for i in range(1, retries + 1):
        try:
            async with attempt_span("external_call", i):
                return await fn()
        except Exception:
            if i == retries:
                raise
            await asyncio.sleep(0.2 * i)

외부 API에서 429 같은 레이트리밋을 다룰 때는 백오프 전략이 특히 중요합니다. 재시도 설계를 더 깊게 보려면 아래 글도 같이 보면 연결이 잘 됩니다.

“리소스 여러 개”를 한 번에 다루는 구성

실무에서는 “DB 세션 + HTTP 세션 + 임시 디렉터리”처럼 여러 리소스를 동시에 엮습니다. 이때 호출부에 async with를 여러 번 쓰는 것도 괜찮지만, 정책을 하나로 묶고 싶을 때가 있습니다.

방법 1: 호출부에서 중첩 async with

async with unit_of_work(session_factory) as db:
    async with http_session() as http:
        ...

가장 명시적이고 디버깅이 쉽습니다.

방법 2: 상위 컨텍스트에서 하위 컨텍스트를 열기

상위 컨텍스트가 하위 리소스를 생성하고, yield로 묶어서 넘깁니다.

from contextlib import asynccontextmanager

@asynccontextmanager
async def app_resources(session_factory):
    async with unit_of_work(session_factory) as db:
        async with http_session() as http:
            yield {"db": db, "http": http}

사용부:

async def handler(session_factory, url: str):
    async with app_resources(session_factory) as r:
        db = r["db"]
        http = r["http"]
        async with http.get(url) as resp:
            ...
        ...

이 방식은 “핸들러가 필요한 리소스 번들”을 표준화할 때 유용합니다.

흔한 함정과 체크리스트

1) yield를 두 번 하면 안 된다

asynccontextmanager는 정확히 한 번 yield해야 합니다. 조건 분기로 인해 yield가 실행되지 않거나 두 번 실행되면 런타임 에러로 이어집니다.

잘못된 예:

@asynccontextmanager
async def bad(flag: bool):
    if flag:
        yield 1
    yield 2

2) 정리 단계에서 예외를 삼키지 말기

정리 단계 finally에서 예외가 발생하면 원래 예외를 덮어버릴 수 있습니다. 정리 로직은 가능한 한 “실패해도 안전”해야 합니다.

  • close가 실패해도 로그만 남기고 지나갈지
  • 아니면 반드시 실패로 처리할지

정책을 명확히 하세요. 보통은 정리 실패를 로깅하고, 원래 예외를 보존하는 방식이 많습니다.

3) 취소(CancelledError)는 특별 취급이 필요할 수 있다

비동기 환경에서는 태스크 취소가 정상 흐름입니다. 파이썬 버전에 따라 CancelledError의 상속 구조가 다를 수 있으니, except Exception으로 롤백을 처리하는 패턴이 “취소도 롤백”이라는 의미가 되는지 확인해야 합니다.

  • 트랜잭션 관점에서는 취소도 롤백이 맞는 경우가 많음
  • 다만 취소는 빠르게 전파돼야 하므로, 정리 로직이 오래 걸리지 않게 해야 함

4) 컨텍스트 내부에서 또 다른 컨텍스트를 열 때는 종료 순서에 주의

리소스 종료 순서는 보통 “열었던 역순”이 안전합니다. async with 중첩을 사용하면 이 규칙이 자동으로 지켜집니다. 수동으로 관리하면 순서가 꼬일 수 있습니다.

언제 이 패턴을 쓰면 좋은가

  • try/finally가 반복되고, 호출부가 장황해지는 순간
  • 성공/실패 시 정책(커밋·롤백, ack·nack, lock release)이 일관돼야 할 때
  • 로깅/메트릭/트레이싱을 “리소스 경계”에 붙이고 싶을 때

반대로, 단순히 객체 하나를 만들고 반환만 하면 되는 경우에는 과도한 추상화가 될 수 있습니다. “정리 책임이 존재하는지”를 기준으로 도입을 판단하는 게 좋습니다.

마무리

contextlib.asynccontextmanager의 3줄 패턴은 비동기 코드에서 가장 자주 반복되는 “생성·사용·정리”를 강제하는 장치입니다. 특히 예외가 일상인 네트워크/DB 환경에서, 이 패턴은 버그를 줄이는 효과가 큽니다.

핵심은 단순합니다.

  • yield 이전에 준비
  • yield로 사용부에 제공
  • finally에서 정리, 필요하면 except에서 정책 처리

이 구조를 팀의 기본 관례로 만들면, 비동기 리소스 누수와 예외 처리 누락을 눈에 띄게 줄일 수 있습니다.

Discuss on Twitter • View on GitHub