Python의 Context Manager와 with 문

1. 개요

파일을 열거나, 데이터베이스에 연결하거나, 락을 유지하는 모든 작업은 마지막에 정리하는 단계가 필요합니다. 이 단계는 중간 코드에서 예외가 발생하더라도 반드시 실행되어야 합니다. Python은 context manager와 with 문으로 이 문제를 해결합니다.

이 글에서는 with가 내부적으로 어떻게 작동하고 class나 contextlib을 사용해서 context manager를 작성하는 방법을 설명합니다. __exit__에서 예외를 처리하는 방법, context manager를 테스트하는 방법, 그리고 사용할 때 흔히 발생하는 실수를 알아보겠습니다.

2. 문제: try/finally로 리소스 정리하기

with를 사용하지 않고 파일을 읽는 방법을 살펴보겠습니다:

f = open("orders.csv", encoding="utf-8")
try:
    header = f.readline()
    print(header)
finally:
    f.close()

이 방법은 맞지만 번거롭습니다. 또한 실수하기 쉽습니다. open()을 try 블록 안에 넣었는데 open()이 실패하면 f가 할당되지 않아서 finally 블록에서 NameError가 발생합니다. with 문은 위의 패턴을 간결하게 만들어줍니다:

with open("orders.csv", encoding="utf-8") as f:
    header = f.readline()
    print(header)

print(f.closed)
order_id,customer,amount

True

with 블록을 빠져나갈 때 파일이 닫혔습니다. 이는 정상적으로 종료되든 예외가 발생하든 동일하게 작동합니다.

3. Context Manager 프로토콜

__enter__와 __exit__ 메서드를 가진 모든 객체는 with와 함께 사용할 수 있습니다. 개념적으로, 다음 문:

with EXPR as target:
    BODY

는 다음 코드와 동등합니다:

manager = EXPR
target = manager.__enter__()
try:
    BODY
except BaseException as e:
    if not manager.__exit__(type(e), e, e.__traceback__):
        raise
else:
    manager.__exit__(None, None, None)

세 가지 중요한 점을 기억하세요:

  • target에 할당되는 값은 EXPR 자체가 아니라 __enter__()의 반환 값입니다.
  • __exit__은 항상 호출됩니다. 예외가 발생했을 때는 (exc_type, exc_value, traceback)을 받습니다. 에러가 없을 때는 세 매개변수 모두 None입니다.
  • __exit__이 truthy 값을 반환하면 예외가 억제됩니다. 그렇지 않으면 예외가 다시 발생합니다.

3.1. class를 사용한 context manager 작성

첫 번째 예제는 코드 블록의 실행 시간을 측정하는 Timer입니다:

import time


class Timer:
    def __init__(self, label: str):
        self.label = label
        self.elapsed: float | None = None

    def __enter__(self):
        self._start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc, tb):
        self.elapsed = time.perf_counter() - self._start
        print(f"[{self.label}] {self.elapsed:.3f}s")
        return False


with Timer("tinh_tong_don_hang") as t:
    total = sum(range(5_000_000))

print(t.elapsed > 0)
[tinh_tong_don_hang] 0.081s
True

각 부분을 설명하면:

  • __enter__는 시작 시간을 기록하고 self를 반환합니다. 덕분에 t는 Timer 객체 자체가 됩니다.
  • __exit__는 경과 시간을 계산하고 출력한 후 False를 반환합니다. 예외가 있으면 억제되지 않습니다.
  • with 블록 후에도 변수 t를 계속 사용할 수 있습니다. with는 새로운 스코프를 생성하지 않기 때문입니다.

3.2. __exit__에서 예외 처리하기

Context manager는 코드 블록이 성공했는지 실패했는지에 따라 다르게 반응해야 할 때 가장 유용합니다. 다음은 ledger(원장)에 대한 간단한 transaction입니다:

class Transaction:
    def __init__(self, ledger: list):
        self.ledger = ledger
        self.pending: list = []

    def __enter__(self):
        return self.pending

    def __exit__(self, exc_type, exc, tb):
        if exc_type is None:
            self.ledger.extend(self.pending)
            print(f"commit {len(self.pending)} 레코드")
        else:
            print(f"{exc_type.__name__}로 인한 롤백: {exc}")
        return False


ledger = []

with Transaction(ledger) as tx:
    tx.append({"order_id": 1, "amount": 250_000})
    tx.append({"order_id": 2, "amount": 120_000})

try:
    with Transaction(ledger) as tx:
        tx.append({"order_id": 3, "amount": 90_000})
        raise ValueError("잔액이 부족합니다")
except ValueError:
    pass

print(len(ledger))
commit 2 레코드
ValueError로 인한 롤백: 잔액이 부족합니다
2

두 번째 with 블록에서 주문 3은 ledger에 기록되지 않았습니다. __exit__이 False를 반환하기 때문에 ValueError는 여전히 발생하고 호출자가 처리 방법을 결정합니다.

주의: __exit__에서 True를 반환하는 것은 예외를 의도적으로 억제하려고 할 때만 해야 합니다. __exit__ 메서드가 실수로 self나 다른 truthy 객체를 반환하면 에러가 흔적도 없이 사라집니다.

4. @contextmanager로 더 간결하게

단순히 setup과 teardown을 위해 전체 class를 작성하는 것은 번거롭습니다. contextlib 모듈의 @contextmanager 데코레이터는 generator를 context manager로 변환합니다:

from contextlib import contextmanager


class PaymentService:
    def connect(self):
        print("결제 포트 연결을 엽니다")

    def charge(self, customer: str, amount: int):
        print(f"{customer}에게서 {amount}을 차감합니다")

    def disconnect(self):
        print("연결을 닫습니다")


@contextmanager
def payment_session(service: PaymentService):
    service.connect()
    try:
        yield service
    finally:
        service.disconnect()


with payment_session(PaymentService()) as payment:
    payment.charge("Lan", 300_000)
결제 포트 연결을 엽니다
Lan에게서 300000을 차감합니다
연결을 닫습니다

데코레이터가 generator를 context manager 프로토콜로 매핑하는 방식:

  • yield 이전 코드는 __enter__ 역할을 합니다.
  • yield된 값은 as 이후 변수에 할당되는 값입니다.
  • yield 이후 코드는 __exit__ 역할을 합니다. with 블록에서 예외가 발생하면 그 예외는 yield 지점에서 다시 발생합니다.

예외가 yield에서 바로 발생하기 때문에, yield 주위의 try/finally 블록은 필수입니다. 없으면 disconnect()는 오류 발생 시 실행되지 않습니다.

각 예외 유형을 다르게 처리하려면 except를 추가하세요:

@contextmanager
def safe_charge(order_id: int):
    try:
        yield
    except ConnectionError as e:
        print(f"주문 {order_id}: 네트워크 오류, 나중에 재시도하겠습니다 ({e})")


with safe_charge(42):
    raise ConnectionError("timeout")

print("계속 처리합니다")
주문 42: 네트워크 오류, 나중에 재시도하겠습니다 (timeout)
계속 처리합니다

Generator는 오류를 잡았지만 다시 발생시키지 않으므로 예외가 억제됩니다. 이 결과는 __exit__이 True를 반환하는 것과 동등합니다.

5. contextlib의 내장 유틸리티

5.1. suppress: 특정 예외 무시하기

import os
from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove("cache_orders.tmp")

print("완료")
완료

위 코드는 try: ... except FileNotFoundError: pass보다 짧고 의도가 명확합니다.

5.2. ExitStack: 미리 알 수 없는 리소스 개수

여러 발주서 파일을 병합해야 하는데 파일 개수는 런타임에만 알 수 있다고 가정하세요:

from contextlib import ExitStack
from pathlib import Path


def merge_invoices(paths: list[Path], output: Path) -> None:
    with ExitStack() as stack:
        files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths]
        with open(output, "w", encoding="utf-8") as out:
            for f in files:
                out.write(f.read())


for i in range(3):
    Path(f"invoice_{i}.txt").write_text(f"발주서 {i}\n", encoding="utf-8")

merge_invoices([Path(f"invoice_{i}.txt") for i in range(3)], Path("all.txt"))
print(Path("all.txt").read_text(encoding="utf-8"))
발주서 0
발주서 1
발주서 2

enter_context는 __enter__를 호출하고 해당 __exit__을 기억합니다. ExitStack이 종료될 때 파일은 역순으로 닫힙니다. 세 번째 파일을 열 때 실패하면 처음 두 파일은 여전히 올바르게 닫힙니다.

5.3. closing과 nullcontext

  • closing(obj)은 close() 메서드만 가진 객체를 context manager로 변환합니다.
  • nullcontext(value)는 아무것도 하지 않는 context manager입니다. 매개변수가 context manager일 수도 있고 아닐 수도 있을 때 유용합니다:
from contextlib import nullcontext
import threading

def update_stock(product: dict, qty: int, lock: threading.Lock | None = None):
    with lock if lock is not None else nullcontext():
        product["stock"] -= qty

product = {"sku": "AO-01", "stock": 10}
update_stock(product, 3)
update_stock(product, 2, threading.Lock())
print(product["stock"])
5

6. 하나의 with 문에서 여러 context manager 사용하기

하나의 with 문에 여러 context manager를 선언할 수 있습니다. Python 3.10 이상에서는 괄호 문법을 사용하여 깔끔하게 줄을 나눌 수 있습니다:

with (
    open("orders.csv", encoding="utf-8") as src,
    open("orders_backup.csv", "w", encoding="utf-8") as dst,
):
    dst.write(src.read())

이 작성 방식은 두 개의 중첩된 with 블록과 동등합니다. src는 먼저 열리고 마지막에 닫힙니다.

7. 비동기 Context Manager

asyncio 코드에서 해당 프로토콜은 __aenter__와 __aexit__이며 async with와 함께 사용됩니다. Generator 버전은 @asynccontextmanager입니다:

import asyncio
from contextlib import asynccontextmanager


@asynccontextmanager
async def order_queue(name: str):
    print(f"queue {name}에 연결합니다")
    await asyncio.sleep(0.01)
    try:
        yield [f"order-{i}" for i in range(3)]
    finally:
        await asyncio.sleep(0.01)
        print(f"queue {name}을 닫습니다")


async def main():
    async with order_queue("payments") as messages:
        for m in messages:
            print("처리", m)

asyncio.run(main())
queue payments에 연결합니다
처리 order-0
처리 order-1
처리 order-2
queue payments을 닫습니다

주의: 일반 with 문을 async context manager와 함께 사용하면 TypeError가 발생합니다 (구형 Python에서는 AttributeError). 반대도 마찬가지입니다. 두 프로토콜은 별개입니다.

8. Context Manager 테스트하기

Context manager는 성공 경로와 오류 경로 모두 테스트해야 합니다. 먼저 pytest를 설치하세요:

pip install pytest

섹션 3.2의 Transaction class를 transaction.py 파일에 넣고 테스트를 작성하세요:

# test_transaction.py
import pytest
from transaction import Transaction


def test_commit_when_no_error():
    ledger = []
    with Transaction(ledger) as tx:
        tx.append({"order_id": 1, "amount": 100})

    assert ledger == [{"order_id": 1, "amount": 100}]


def test_rollback_and_propagate_error():
    ledger = []
    with pytest.raises(ValueError, match="잔액"):
        with Transaction(ledger) as tx:
            tx.append({"order_id": 2, "amount": 999})
            raise ValueError("잔액이 부족합니다")

    assert ledger == []
pytest -q
..                                                       [100%]
2 passed in 0.01s

두 번째 테스트는 두 가지를 동시에 확인합니다: 데이터가 롤백되고, 예외가 억제되지 않습니다. pytest.raises 자체도 context manager이며, 이는 이 글의 주제의 훌륭한 실제 예입니다.

9. 흔한 실수와 사용하면 안 되는 경우

yield 주위에 try/finally를 빠뜨렸을 때. 이는 @contextmanager 사용 시 가장 흔한 실수입니다:

@contextmanager
def broken_session(service):
    service.connect()
    yield service
    service.disconnect()  # with 블록에서 예외 발생 시 실행되지 않습니다

with 종료 후 리소스를 사용할 때. 변수는 계속 존재하지만 리소스는 이미 해제되었습니다:

with open("orders.csv", encoding="utf-8") as f:
    pass
f.read()
ValueError: I/O operation on closed file.

Generator 내에서 여러 번 yield할 때. @contextmanager와 함께 사용되는 generator는 정확히 한 번만 yield해야 합니다. 더 많이 yield하면 Python은 RuntimeError: generator didn't stop를 발생시킵니다.

@contextmanager에서 생성된 context manager를 재사용할 때. 이 객체는 한 번만 사용할 수 있습니다. 같은 객체에서 두 번째 with를 호출하면 오류가 발생합니다. 매번 사용할 때마다 함수를 호출하여 새 객체를 생성하세요.

남용. context manager를 다음 경우에는 사용하지 말아야 합니다:

  • 리소스가 단일 코드 블록보다 더 오래 살아야 하는 경우, 예를 들어 애플리케이션 전체 수명 동안 존재하는 connection pool. 이 경우 애플리케이션 시작/종료 계층에서 관리하세요.
  • 명확한 “열기/닫기” 쌍이 없는 로직. 일반 함수가 더 읽기 쉽습니다.
  • suppress(Exception)으로 모든 예외를 무시하려고 할 때. 이 방식은 except: pass처럼 버그를 숨깁니다.

주의: 리소스를 닫기 위해 garbage collector나 __del__에 의존하지 마세요. 이들이 실행되는 시점은 보장되지 않으며, 특히 PyPy 같은 구현에서는 더욱 그렇습니다. with를 사용하세요.

10. 결론

Context manager는 setup과 teardown을 한곳에 묶고 예외가 발생해도 teardown이 항상 실행되도록 보장합니다. 다음을 배웠습니다:

  • __enter__와 __exit__ 프로토콜, __exit__ 반환 값의 의미.
  • @contextmanager를 사용한 간결한 작성, yield 주위의 try/finally가 필수인 이유.
  • suppress, ExitStack, closing, nullcontext와 비동기 버전.
  • pytest로 commit 경로와 rollback 경로 모두 테스트하는 방법.

코드에서 “가져온 후 반환” 쌍이 있을 때마다 (파일, 락, transaction, 연결 등), context manager로 감싸는 것을 고려하세요.

pythoncontext-managercontextlibresource-management