Python のコンテキストマネージャ(with文)を自作する
Python でファイルを扱うときに書く with open(...) as f: の with 文。この裏側で動いているのが「コンテキストマネージャ」です。本記事では、その仕組みを理解し、クラスと contextlib の2通りで自作する方法を入門者向けに解説します。
対象読者は「with open() は使えるが、自分で with に対応したオブジェクトを作ったことはない」という方です。環境は Python 3.8 以上(コードは 3.12 で確認)、追加ライブラリは不要です。
コンテキストマネージャとは何か
コンテキストマネージャとは、with 文で使えるオブジェクトのことです。やっていることは2つだけです。
- ブロックに入るときに「前処理(セットアップ)」をする
- ブロックを抜けるときに「後処理(クリーンアップ)」をする
重要なのは、ブロック内で例外が発生しても後処理が必ず実行されるという点です。
ファイルを例に考えます。
f = open("data.txt")
try:
data = f.read()
finally:
f.close() # 例外が起きても必ず閉じる
この try / finally を毎回書くのは面倒で、close() の書き忘れも起こります。with 文はこの定型を隠蔽します。
with open("data.txt") as f:
data = f.read()
# ブロックを抜けた時点で f は必ず閉じられている
この「必ず後処理する」をユーザー定義のオブジェクトでも実現できるのがコンテキストマネージャです。
仕組み:__enter__ と __exit__
with 文に対応するオブジェクトは、次の2つの特殊メソッドを持ちます。
__enter__(self)— ブロックに入るときに呼ばれる。戻り値がasの変数に入る__exit__(self, exc_type, exc_value, traceback)— ブロックを抜けるときに呼ばれる。後処理を書く
with cm as x: は、おおよそ次と等価です。
x = cm.__enter__()
try:
... # with ブロックの中身
finally:
cm.__exit__(例外の型, 例外の値, トレースバック)
例外が起きなかった場合、__exit__ の3引数はすべて None になります。
方法1:クラスで自作する
一番基本的な方法が、__enter__ / __exit__ を持つクラスを定義することです。処理時間を計測するコンテキストマネージャを作ってみます。
import time
class Timer:
def __init__(self, label="処理"):
self.label = label
def __enter__(self):
self.start = time.perf_counter()
return self # as で受け取る値
def __exit__(self, exc_type, exc_value, traceback):
elapsed = time.perf_counter() - self.start
print(f"{self.label}: {elapsed:.4f} 秒")
with Timer("集計処理"):
total = sum(range(10_000_000))
# 出力例: 集計処理: 0.1532 秒
__enter__ が return self しているので、as t: と書けば Timer インスタンス自体を受け取れます。
例外が起きても後処理は走る
コンテキストマネージャの本質は「例外が起きても後処理が走る」ことです。
class Resource:
def __enter__(self):
print("→ 接続を開く")
return self
def __exit__(self, exc_type, exc_value, traceback):
print("→ 接続を閉じる")
with Resource():
print("処理中")
raise ValueError("想定外のエラー")
実行結果は次のようになります。
→ 接続を開く 処理中 → 接続を閉じる Traceback (most recent call last): ... ValueError: 想定外のエラー
raise で中断されても「接続を閉じる」が実行されています。これが DB 接続・ファイル・ロック・一時ディレクトリなど「必ず解放したいリソース」の管理に向く理由です。
__exit__ の戻り値で例外を抑制する
__exit__ が True を返すと、ブロック内で起きた例外を抑制できます。
class SuppressValueError:
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
return exc_type is ValueError # ValueError だけ握りつぶす
with SuppressValueError():
raise ValueError("これは無視される")
print("ここまで到達する")
便利ですが、例外を安易に握りつぶすとバグの温床になります。デフォルト(None を返す)では例外はそのまま伝播します。
方法2:contextlib.contextmanager で自作する
クラスが大げさに感じる場合、contextlib.contextmanager デコレータを使うとジェネレータ関数1つで書けます。
import time
from contextlib import contextmanager
@contextmanager
def timer(label="処理"):
start = time.perf_counter()
try:
yield # ここで with ブロックに制御が渡る
finally:
elapsed = time.perf_counter() - start
print(f"{label}: {elapsed:.4f} 秒")
with timer("集計処理"):
total = sum(range(10_000_000))
ポイントは yield です。
yieldの 前 が__enter__相当(セットアップ)yieldでwithブロックに制御が渡るyieldの 後(finally)が__exit__相当(クリーンアップ)
値を as で渡したい場合は yield value と書きます。
@contextmanager
def opened(path):
f = open(path)
try:
yield f # as f: でこの値を受け取る
finally:
f.close()
with opened("data.txt") as f:
print(f.read())
例外に備えて、yield は必ず try / finally で囲むのが鉄則です。囲まないと、ブロック内で例外が起きたときにクリーンアップが走りません。
クラス版とデコレータ版の使い分け
| 観点 | クラス版 | @contextmanager 版 |
|---|---|---|
| 記述量 | 多め | 少ない |
| 状態を複数メソッドで共有 | しやすい | しにくい |
| 再利用(複数回 with する) | インスタンス次第で可 | 原則1回 |
| 単純な前後処理の読みやすさ | やや冗長 | 直感的 |
- 単純な「前処理 → yield → 後処理」なら
@contextmanager - 状態やメソッドを持たせたい、使い回したいならクラス
注意点
@contextmanager版でyieldをtry / finallyで囲まないと、例外時にクリーンアップが走りません__exit__/finally内でさらに例外を投げると、元の例外が隠れることがあります- ジェネレータベースのコンテキストマネージャは複数回
withで使い回せません - 動的な個数のリソースを管理したい場合は
contextlib.ExitStackが便利です
from contextlib import ExitStack
paths = ["a.txt", "b.txt", "c.txt"]
with ExitStack() as stack:
files = [stack.enter_context(open(p)) for p in paths]
# ブロックを抜ければ全ファイルが閉じられる
まとめ
with文の裏側はコンテキストマネージャ(__enter__/__exit__)- 本質は「例外が起きても後処理を必ず実行する」こと
- 作り方は2通り:クラス定義、または
@contextmanager+ ジェネレータ - 単純な前後処理はデコレータ版、状態を持たせたいならクラス版
「try / finally を毎回書いている」コードを見つけたら、コンテキストマネージャに切り出せないか検討してみてください。意図が明確になり、解放漏れも防げます。
