モダンPython中級 #3 コンテキストマネージャー (with、contextlib)

プログラミング言語 Python context manager with
読了 6分

基礎 #6 で try/finally よりも with の方がすっきりすると軽く触れて通り過ぎました。今回がそのテーマです。with のあらゆるパターン と、ユーザー定義のコンテキストマネージャー、contextlib のツールを扱います。

with — 一行で片付くリソース #

最も基本的なところ — ファイル処理。

with なしのとき 
f = open("data.txt")
try:
    data = f.read()
finally:
    f.close()
with と一緒に 
with open("data.txt") as f:
    data = f.read()
# ここで自動的に f.close()

with は ブロックを抜けるときに後片付けコードを自動で呼んでくれます。 正常終了、例外発生、return どの経路で抜けても後片付けは行われます。

with が合うリソース #

リソース後片付け動作
ファイルclose()
ロック (Lock)release()
DB 接続 / トランザクションcommit() または rollback() + close()
一時ディレクトリディレクトリ削除
標準出力のリダイレクト元の stdout に復帰
環境変数の変更元の値に復帰

共通点: 「一時的にある状態に入って、抜けるときに元に戻したい」。これがコンテキストマネージャーの出番です。

複数のリソースを一度に #

複数の with — カンマで 
with open("input.txt") as src, open("output.txt", "w") as dst:
    dst.write(src.read())

3.10 からは 括弧で囲んで複数行で書く こともできます。

複数行の with (3.10+) 
with (
    open("input.txt") as src,
    open("output.txt", "w") as dst,
    Lock() as lock,
):
    dst.write(src.read())

引数の多い関数と同じ形なので読みやすいです。

自分で作る — __enter__ / __exit__ #

with で動作するオブジェクトは 2 つのメソッドを持ちます。

  • __enter__(self) — with 進入時に呼ばれ、as の後ろで受け取る値を返す
  • __exit__(self, exc_type, exc_value, traceback) — 抜けるときに呼ばれる

最も単純な例 — 一時的な chdir #

作業ディレクトリを一時的に変えたいとき。

リソースクラスを作る 
import os

class chdir:
    def __init__(self, path: str):
        self.path = path

    def __enter__(self):
        self.old = os.getcwd()
        os.chdir(self.path)
        return self.path

    def __exit__(self, exc_type, exc_value, tb):
        os.chdir(self.old)

with chdir("/tmp"):
    print(os.getcwd())     # /tmp
print(os.getcwd())          # 元のディレクトリ

__exit__ の戻り値 — 例外を吸収するか #

__exit__ は truthy を返すと例外を飲み込みます。 これを知らずに True を返すとすべての例外が消えてデバッグしにくくなります。

🚫 意図しない例外の吸収 
class Bad:
    def __enter__(self): return self
    def __exit__(self, *args):
        return True   # ✗ すべての例外を飲み込む

ほとんどの場合、何も返さないか False / None にしておきましょう。

特定の例外だけ意図的に吸収したいときは exc_type を見て判断します。

特定の例外だけ吸収 
class IgnoreFileNotFound:
    def __enter__(self): return self
    def __exit__(self, exc_type, exc_value, tb):
        return exc_type is not None and issubclass(exc_type, FileNotFoundError)

with IgnoreFileNotFound():
    with open("missing.txt") as f:
        ...
# 例外なしで通過

このようなよくあるパターンは標準ライブラリにすでにあります (下の suppress 参照)。

@contextmanager — 関数ひとつで短く #

__enter__ / __exit__ の 2 メソッドを持つクラスを毎回作るのは面倒です。関数ひとつに yield を挟んで コンテキストマネージャーを作る方法があります。

@contextmanager 
from contextlib import contextmanager

@contextmanager
def chdir(path: str):
    old = os.getcwd()
    os.chdir(path)
    try:
        yield path        # ここで with ブロックが実行される
    finally:
        os.chdir(old)

この一つの関数が上のクラス版とまったく同じことをします。

読み方:

  • yield の 前 = __enter__ がやること
  • yield の 値 = as の後ろで受け取る値
  • yield の 後 = __exit__ がやること
  • try/finally で囲まないと例外時の後片付けが保証されない

例外も受け取りたいなら #

例外を捕まえる 
@contextmanager
def transactional(conn):
    tx = conn.begin()
    try:
        yield tx
    except Exception:
        tx.rollback()
        raise
    else:
        tx.commit()

yield のところで with ブロックが例外を投げると、それがジェネレータの yield のところで再び投げられます。だから普通の try/except で捕まえられます。

contextlib の他の補助ツール #

suppress — 特定の例外を無視 #

suppress 
from contextlib import suppress
import os

with suppress(FileNotFoundError):
    os.remove("maybe-missing.txt")
# ファイルがなくてもそのまま通過

上で作った IgnoreFileNotFound クラスがそのまま標準にあります。

closing — close() メソッドだけ持つオブジェクトを包む #

closing 
from contextlib import closing
from urllib.request import urlopen

with closing(urlopen("https://example.com")) as resp:
    data = resp.read()
# resp.close() が自動

urlopen のレスポンスオブジェクトが __exit__ を持っていなかった旧バージョンでよく使われていました。最近はレスポンスオブジェクトも with を直接サポートする場合が多いです。

redirect_stdout — 標準出力を横取り #

redirect_stdout 
import io
from contextlib import redirect_stdout

buf = io.StringIO()
with redirect_stdout(buf):
    print("hello")
print("captured:", buf.getvalue())
# captured: hello

テスト、キャプチャ、ログ変換のような場面で有用です。redirect_stderr も同じ動作です。

nullcontext — 条件付きで with を使う #

nullcontext 
from contextlib import nullcontext

def process(path: str | None):
    ctx = open(path) if path else nullcontext()
    with ctx as f:
        if f is None:
            print("no file")
        else:
            print(f.read())

「ファイルがあれば開いて、なければそのまま通す」のようなケース。with のところには常にコンテキストマネージャーが必要なので、空きを埋める無害なオブジェクト が必要です。nullcontext がそれです。

ExitStack — 動的に複数のリソースを管理 #

リソースの個数が ランタイムで決まる とき。

ExitStack 
from contextlib import ExitStack

def merge(paths: list[str]) -> str:
    with ExitStack() as stack:
        files = [stack.enter_context(open(p)) for p in paths]
        return "\n".join(f.read() for f in files)
# すべてのファイルが自動で close

with open(p) as f を N 個書けないとき (paths のリスト長が動的なら) ExitStack がそれを解いてくれます。登録されたすべてのリソースが逆順に後片付け されます。

ExitStack で __enter__ ではない一般の後片付け関数も登録できます。

コールバックの登録 
def setup_temp_resources():
    stack = ExitStack()
    tmp_dir = create_tmp_dir()
    stack.callback(remove_dir, tmp_dir)
    # ... さらに多くのリソースを登録
    return stack

with setup_temp_resources() as stack:
    ...

async コンテキストマネージャー — 軽くプレビュー #

非同期リソースには async with を使います。

async with 
async with open_async_db() as conn:
    await conn.execute(...)

これはメソッドが __aenter__ / __aexit__ です。@asynccontextmanager もあり、使い方は同期版と同じです。詳しくは #7 非同期入門 で。

ユーザー定義 — クラス vs @contextmanager #

ケースクラス@contextmanager
短くて単純⭕✅
状態 / メソッドが多い✅❌
例外処理の分岐が複雑✅⭕
短い副作用 (chdir、環境変数など)⭕✅
ライブラリでユーザーに公開✅⭕

迷ったら @contextmanager から 試して、複雑になったらクラスに移すのが無難です。

よく出会うパターン — 一ヶ所に #

一時的な環境変数 #

環境変数の一時設定 
import os
from contextlib import contextmanager

@contextmanager
def set_env(**kwargs: str):
    old = {}
    for k, v in kwargs.items():
        old[k] = os.environ.get(k)
        os.environ[k] = v
    try:
        yield
    finally:
        for k, v in old.items():
            if v is None:
                del os.environ[k]
            else:
                os.environ[k] = v

テストで環境変数に依存するコードを検証するときによく使います。

時間計測 #

elapsed time 
import time
from contextlib import contextmanager

@contextmanager
def timer(label: str):
    start = time.perf_counter()
    yield
    elapsed = time.perf_counter() - start
    print(f"{label}: {elapsed:.3f}s")

with timer("query"):
    do_expensive_thing()
# query: 0.123s

ロック #

lock 
import threading

lock = threading.Lock()

with lock:
    # クリティカルセクション
    update_shared_state()

Lock、RLock、Semaphore すべて with をサポートします。acquire() / release() を直接呼ぶコードはほとんど書きません。

まとめ #

今回押さえたツール:

  • with — リソースの出入りを一行で片付け、例外時も後片付けを保証
  • 複数リソース — with A, B: または括弧で複数行
  • 自分で作る: __enter__ / __exit__、__exit__ の戻り値は False が一般的
  • @contextmanager — 関数ひとつで短く作る標準ツール、yield の前後 = 進入 / 後片付け
  • contextlib 補助: suppress、closing、redirect_stdout、nullcontext、ExitStack
  • ExitStack — リソース個数が動的なとき、コールバックの登録も可能
  • 短い副作用 (chdir、env、timer) は @contextmanager が最も合う場面
  • 非同期は async with + @asynccontextmanager

次回 (#4 イテラブル / ジェネレータ / yield from) では、内包表記の先のツール — ユーザー定義のイテラブル、ジェネレータ関数、yield from を扱います。上で見た @contextmanager も実はジェネレータの上に作られたツールです。

X