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__の2つのメソッドを持つオブジェクトはすべて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)

3つのポイントを覚えておきましょう。

• targetに代入される値は**__enter__()の戻り値**であり、EXPRそのものではありません。
• __exit__は常に呼び出されます。例外がある場合は(exc_type, exc_value, traceback)を受け取ります。エラーがない場合、3つのパラメータはすべてNoneです。
• __exit__が真の値を返すと、例外は抑制されます。そうでない場合、例外は再度発生します。

3.1. クラスを使って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("合計注文を計算") as t:
    total = sum(range(5_000_000))

print(t.elapsed > 0)

[合計注文を計算] 0.081s
True

各部分の説明：

• __enter__は開始時刻を記録し、selfを返します。そのためtはTimerオブジェクトそのものになります。
• __exit__は経過時間を計算して表示し、Falseを返します。例外があっても抑制されません。
• 変数tはwithブロック後も使用できます。理由はwithが新しいスコープを作成しないためです。

3.2. __exit__での例外処理

Context managerが最も活躍するのは、コードブロックの成功・失敗に応じて異なる処理が必要な場合です。以下の例は帳簿(ledger)上の単純なトランザクションです。

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"コミット {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))

コミット 2 レコード
ロールバック（ValueError: 残高不足です）
2

2番目のwithブロックでは、オーダー3はledgerに記録されていません。__exit__がFalseを返すため、ValueErrorはそのまま発生し、呼び出し側で処理できます。

注意: __exit__からTrueを返すのは、意図的に例外を抑制したい場合だけです。誤ってreturn selfまたは他の真の値を返すと、エラーが無音で抑制されてしまいます。

4. @contextmanagerでより簡潔に

setupとteardownの1ステップだけのためにクラス全体を書くのはかなり冗長です。モジュールcontextlibは、generatorをcontext managerに変換するデコレータ@contextmanagerを提供します。

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が終了すると、ファイルはオープンした時とは逆の順序で閉じられます。3番目のファイルのオープンが失敗した場合でも、最初の2つのファイルは正しく閉じられます。

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())

このコードは2つのネストされたwithブロックと同等です。srcは最初に開かれ、最後に閉じられます。

7. Async context manager

asyncioコードでは、対応するプロトコルは__aenter__と__aexit__で、async withで使用します。そのgenerator版は@asynccontextmanagerです。

import asyncio
from contextlib import asynccontextmanager


@asynccontextmanager
async def order_queue(name: str):
    print(f"キュー{name}に接続します")
    await asyncio.sleep(0.01)
    try:
        yield [f"order-{i}" for i in range(3)]
    finally:
        await asyncio.sleep(0.01)
        print(f"キュー{name}を切断します")


async def main():
    async with order_queue("payments") as messages:
        for m in messages:
            print("処理中:", m)

asyncio.run(main())

キューpaymentsに接続します
処理中: order-0
処理中: order-1
処理中: order-2
キューpaymentsを切断します

注意: 通常のwithをasync context managerと一緒に使うとTypeErrorが発生します（古いPythonバージョンではAttributeError）。その逆も同じです。これら2つのプロトコルは分離されています。

8. Context Managerのテスト

context managerは成功パスとエラーパスの両方でテストされるべきです。まずpytestをインストールしてください。

pip install pytest

セクション3.2のTransactionクラスを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

2番目のテストは2つのことを同時にチェックします：データがロールバックされ、例外が抑制されないこと。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は正確に1回yieldする必要があります。複数回yieldすると、PythonはRuntimeError: generator didn't stopを発生させます。

@contextmanagerから作成されたcontext managerを再利用する. このオブジェクトは1回だけ使用できます。同じオブジェクト上で2番目のwithを呼び出すとエラーになります。毎回、関数を呼び出して新しいオブジェクトを作成してください。

過剰な使用. 以下の場合はcontext managerを使うべきではありません。

• リソースが単一のコードブロックより長く存在する必要がある場合。例えば、アプリケーションのライフサイクル全体にわたるconnection pool。その場合は、アプリケーションの起動とシャットダウンのレベルで管理してください。
• “開く/閉じる”の明確なペアがないロジック。通常の関数の方が読みやすいでしょう。
• suppress(Exception)ですべての例外を抑制したい場合。これはexcept: passと同じようにバグを隠してしまいます。

注意: リソースを閉じるためにガベージコレクタまたは__del__に依存しないでください。それらの実行時刻は保証されず、特にPyPyのような実装では不確実です。withを使用してください。

10. まとめ

context managerはsetupとteardownを1つの場所に集約し、例外が発生した場合でもteardownが常に実行されることを保証します。以下を学習しました。

• __enter__と__exit__プロトコル、および__exit__からの戻り値の意味。
• @contextmanagerでの簡潔な記述方法、およびyieldの周りのtry/finallyの必須性。
• ユーティリティsuppress、ExitStack、closing、nullcontext、およびそれらのasync版。
• pytestを使ったcommitとrollbackの両方のブランチのテスト方法。

コードに“取得してから解放する”ペア（ファイル、ロック、トランザクション、接続など）がある場合は、context managerでラップすることを検討してください。