2026年5月30日
2026年5月30日
Pythonのcontextmanagerの使い方|with文で後片付けを自動化する
Pythonの contextlib.contextmanager を使うと、「開始処理 → 本処理 → 後片付け」という型を、ジェネレーター関数1つと with 文で書けるようになります。この記事では、with 文の仕組みから @contextmanager の書き方、例外が起きたときの挙動、標準ライブラリにある便利なコンテキストマネージャー、よくあるつまずきまでを、サーバー運用スクリプトでそのまま使える例で解説します。
コード例は(2026年10月現在)サポートが続いている Python 3 系を前提にしています。バージョン差があるものは本文で注記します。
with文が解決する問題
ファイルやロック、DB接続、一時ディレクトリのように「使ったら戻す」必要がある資源は、途中で例外が起きても後片付けを実行しなければなりません。try / finally で書くこともできますが、同じ型の処理が増えるとコードが読みにくくなり、片付け漏れも起きやすくなります。
f = open("access.log", encoding="utf-8")
try:
for line in f:
print(line.rstrip())
finally:
f.close()
with 文はこの型を言語として用意したものです。
with open("access.log", encoding="utf-8") as f:
for line in f:
print(line.rstrip())
# ブロックを抜けた時点で、例外の有無にかかわらず f は閉じられている
with文の裏で呼ばれているもの
with 文に渡すオブジェクト(コンテキストマネージャー)は、2つの特殊メソッドを持っています。
__enter__():ブロックに入るときに呼ばれ、戻り値がasの変数に入る__exit__(exc_type, exc_value, traceback):ブロックを抜けるときに呼ばれる。例外がなければ3つの引数はすべてNone
__exit__ が真の値を返すと、ブロック内で起きた例外はそこで握りつぶされます。意図せず真を返すと障害の原因が見えなくなるため、通常は None(何も返さない)にしておきます。
クラスで書くと次のようになります。
import time
class Timer:
def __init__(self, label):
self.label = label
def __enter__(self):
self.start = time.perf_counter()
return self
def __exit__(self, exc_type, exc_value, traceback):
elapsed = time.perf_counter() - self.start
print(f"{self.label}: {elapsed:.3f}秒")
# 何も返さない = 例外があればそのまま外へ伝わる
with Timer("集計"):
total = sum(range(10_000_000))
@contextmanagerで関数として書く
クラスを書くほどでもない場合は、contextlib.contextmanager デコレーターを使います。ジェネレーター関数の yield の前が __enter__、後ろが __exit__ に相当し、yield した値が as の変数に入ります。
import time
from contextlib import contextmanager
@contextmanager
def timer(label):
start = time.perf_counter()
try:
yield
finally:
elapsed = time.perf_counter() - start
print(f"{label}: {elapsed:.3f}秒")
with timer("バックアップ圧縮"):
time.sleep(1.2)
ここで重要なのが try / finally です。with ブロック内で例外が起きると、その例外は yield の位置で関数の中に投げ込まれます。finally がないと yield より後ろの片付けコードは実行されません。@contextmanager を書くときは「yield を try で囲む」を基本形として覚えておくと事故を防げます。
値を渡す・状態を一時的に変える
運用スクリプトでよく使うのが「一時的に設定を変えて、終わったら元に戻す」型です。環境変数を一時的に差し替える例を示します。
import os
from contextlib import contextmanager
@contextmanager
def temp_env(**overrides):
saved = {key: os.environ.get(key) for key in overrides}
os.environ.update(overrides)
try:
yield dict(os.environ)
finally:
for key, old in saved.items():
if old is None:
os.environ.pop(key, None)
else:
os.environ[key] = old
with temp_env(LANG="C", TZ="UTC") as env:
print(env["TZ"]) # UTC
# ブロックを抜けると元の値に戻る
元の値が「未設定」だった場合は削除して戻す、という分岐を入れておくのがポイントです。単純に保存した値を代入するだけだと、存在しなかった変数が空文字や None 扱いで残ってしまいます。
例外の扱いを設計する
@contextmanager の中で例外を捕まえて再送出しなければ、その例外は抑止されます。ログに残してから外へ伝えたい場合は raise で再送出します。
import logging
from contextlib import contextmanager
log = logging.getLogger(__name__)
@contextmanager
def logged_step(name):
log.info("開始: %s", name)
try:
yield
except Exception:
log.exception("失敗: %s", name)
raise # 再送出しないと呼び出し元は失敗に気づけない
else:
log.info("完了: %s", name)
「失敗しても処理を続けたい」場面では、例外を握りつぶすより、対象の例外クラスを明示する contextlib.suppress を使う方が意図が伝わります。
import os
from contextlib import suppress
with suppress(FileNotFoundError):
os.remove("/tmp/old.lock") # 無ければ何もしない
except Exception: pass のように広く握りつぶすと、権限エラーやディスク不足まで見えなくなります。抑止する例外は必要最小限に絞ってください。
標準ライブラリの便利なコンテキストマネージャー
自作する前に、標準ライブラリに同じものがないか確認すると手間が減ります。
tempfile.TemporaryDirectory():一時ディレクトリを作り、ブロックを抜けると中身ごと削除するcontextlib.suppress(*exceptions):指定した例外だけを無視するcontextlib.closing(obj):close()を持つがwithに対応していないオブジェクトを閉じるcontextlib.redirect_stdout(f):標準出力の行き先を一時的に変えるcontextlib.nullcontext():「何もしない」コンテキストマネージャー。条件によってロックを使う/使わないを切り替えるときに便利(Python 3.7以降)contextlib.chdir(path):カレントディレクトリを一時的に変える(Python 3.11以降)。プロセス全体の状態を変えるため、スレッドを使うプログラムでは注意が必要ですcontextlib.ExitStack():個数が実行時に決まる複数のコンテキストマネージャーをまとめて管理する
ExitStack は、たとえば引数で渡された複数のログファイルをすべて開き、最後にまとめて閉じたいときに使います。
import sys
from contextlib import ExitStack
paths = sys.argv[1:]
with ExitStack() as stack:
files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths]
for f in files:
print(f.name, sum(1 for _ in f))
# 開いた順と逆の順ですべて閉じられる
途中のファイルで open() が失敗しても、それまでに開いたファイルは閉じられます。リスト内包表記で open() を並べるだけでは、この保証は得られません。
複数を1つのwith文に並べる
複数のコンテキストマネージャーはカンマで並べられます。Python 3.10以降は括弧で囲んで複数行に分けて書く形が正式にサポートされています。
with (
open("in.csv", encoding="utf-8") as src,
open("out.csv", "w", encoding="utf-8") as dst,
):
for line in src:
dst.write(line.upper())
非同期処理ではasynccontextmanager
asyncio を使うコードでは、async with に対応した contextlib.asynccontextmanager(Python 3.7以降)を使います。書き方は同期版と同じで、関数を async def にし、片付け処理で await できます。
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def limited(sem):
await sem.acquire()
try:
yield
finally:
sem.release()
async def main():
sem = asyncio.Semaphore(3)
async with limited(sem):
await asyncio.sleep(0.1)
asyncio.run(main())
よくあるつまずき
yieldを2回書いた/書き忘れた
@contextmanager の関数は yield をちょうど1回だけ実行する必要があります。条件分岐で yield に到達しないと RuntimeError(generator didn't yield)、2回到達すると同じく RuntimeError(generator didn't stop)になります。分岐が多い場合は、yield を関数の中央に1つだけ置き、分岐はその前後で処理する構成にすると間違えにくくなります。
sqlite3のwithは接続を閉じない
sqlite3 の接続オブジェクトを with に渡すと、ブロックの成否に応じてコミットまたはロールバックが行われますが、接続そのものは閉じられません。接続も閉じたい場合は contextlib.closing と組み合わせます。
import sqlite3
from contextlib import closing
with closing(sqlite3.connect("app.db")) as conn:
with conn: # トランザクション: 成功でcommit、例外でrollback
conn.execute("UPDATE jobs SET status = ? WHERE id = ?", ("done", 42))
ライブラリごとに with が何をするか(閉じるのか、コミットするだけか)は異なります。使う前に公式ドキュメントで確認してください。
ジェネレーターを使い回そうとする
@contextmanager で作った関数は、呼び出すたびに新しいコンテキストマネージャーを返します。cm = timer("x") として作った同じオブジェクトを2回 with に渡すことはできません。毎回 with timer("x"): のように呼び出してください。なお、@contextmanager で作ったものはデコレーターとしても使え、その場合は関数が呼ばれるたびに新しく作り直されます。
まとめ
with文は__enter__と__exit__を呼ぶ仕組みで、例外が起きても後片付けを実行できる@contextmanagerではyieldの前が開始処理、後ろが片付け。yieldはtry/finallyで囲むのが基本形- 例外を抑止したいときは
suppressで対象を絞り、ログに残す場合は再送出を忘れない TemporaryDirectory・closing・ExitStack・nullcontextなど、標準ライブラリの部品を先に探すyieldは1回だけ。sqlite3のように「閉じないwith」もあるため、ライブラリの挙動は公式ドキュメントで確認する