コンテキストマネージャ (with、contextlib)

f = open("data.txt")
try:
    data = f.read()
finally:
    f.close()

with open("data.txt") as f:
    data = f.read()
# ここで自動的に f.close()

with open("input.txt") as src, open("output.txt", "w") as dst:
    dst.write(src.read())

with (
    open("input.txt") as src,
    open("output.txt", "w") as dst,
    Lock() as lock,
):
    dst.write(src.read())

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())          # 元のディレクトリ

class Bad:
    def __enter__(self): return self
    def __exit__(self, *args):
        return True   # ✗ すべての例外を飲み込む

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:
        ...
# 例外なく通過

from contextlib import contextmanager

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

@contextmanager
def transactional(conn):
    tx = conn.begin()
    try:
        yield tx
    except Exception:
        tx.rollback()
        raise
    else:
        tx.commit()

from contextlib import suppress
import os

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

from contextlib import closing
from urllib.request import urlopen

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

import io
from contextlib import redirect_stdout

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

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

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

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 with open_async_db() as conn:
    await conn.execute(...)

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

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

import threading

lock = threading.Lock()

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