python-resource-management

Python のリソース管理

コンテキストマネージャーを使って、リソースを確実に管理します。データベース接続、ファイルハンドル、ネットワークソケットなどのリソースは、例外が発生した場合でも確実に解放する必要があります。

このスキルを使う場面

• データベース接続や接続プールを管理する
• ファイルハンドルや I/O を扱う
• カスタムコンテキストマネージャーを実装する
• 状態を持つストリーミングレスポンスを構築する
• ネストしたリソースのクリーンアップを処理する
• 非同期コンテキストマネージャーを作成する

主な概念

1. コンテキストマネージャー

with 文を使うと、例外が発生した場合でもリソースが自動的に解放されます。

2. プロトコルメソッド

同期処理では __enter__／__exit__、非同期処理では __aenter__／__aexit__ を使ってリソースを管理します。

3. 無条件のクリーンアップ

例外が発生したかどうかにかかわらず、__exit__ は必ず実行されます。

4. 例外処理

例外を抑制するには __exit__ から True を返し、伝播させるには False を返します。

クイックスタート

from contextlib import contextmanager

@contextmanager
def managed_resource():
    resource = acquire_resource()
    try:
        yield resource
    finally:
        resource.cleanup()

with managed_resource() as r:
    r.do_work()

基本パターン

パターン 1：クラスベースのコンテキストマネージャー

複雑なリソースに対して、コンテキストマネージャープロトコルを実装します。

class DatabaseConnection:
    """Database connection with automatic cleanup."""

    def __init__(self, dsn: str) -> None:
        self._dsn = dsn
        self._conn: Connection | None = None

    def connect(self) -> None:
        """Establish database connection."""
        self._conn = psycopg.connect(self._dsn)

    def close(self) -> None:
        """Close connection if open."""
        if self._conn is not None:
            self._conn.close()
            self._conn = None

    def __enter__(self) -> "DatabaseConnection":
        """Enter context: connect and return self."""
        self.connect()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Exit context: always close connection."""
        self.close()

# Usage with context manager (preferred)
with DatabaseConnection(dsn) as db:
    result = db.execute(query)

# Manual management when needed
db = DatabaseConnection(dsn)
db.connect()
try:
    result = db.execute(query)
finally:
    db.close()

パターン 2：非同期コンテキストマネージャー

非同期リソースには、非同期プロトコルを実装します。

class AsyncDatabasePool:
    """Async database connection pool."""

    def __init__(self, dsn: str, min_size: int = 1, max_size: int = 10) -> None:
        self._dsn = dsn
        self._min_size = min_size
        self._max_size = max_size
        self._pool: asyncpg.Pool | None = None

    async def __aenter__(self) -> "AsyncDatabasePool":
        """Create connection pool."""
        self._pool = await asyncpg.create_pool(
            self._dsn,
            min_size=self._min_size,
            max_size=self._max_size,
        )
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Close all connections in pool."""
        if self._pool is not None:
            await self._pool.close()

    async def execute(self, query: str, *args) -> list[dict]:
        """Execute query using pooled connection."""
        async with self._pool.acquire() as conn:
            return await conn.fetch(query, *args)

# Usage
async with AsyncDatabasePool(dsn) as pool:
    users = await pool.execute("SELECT * FROM users WHERE active = $1", True)

パターン 3：@contextmanager デコレーターの使用

単純なケースでは、デコレーターを使ってコンテキストマネージャーを簡潔に記述します。

from contextlib import contextmanager, asynccontextmanager
import time
import structlog

logger = structlog.get_logger()

@contextmanager
def timed_block(name: str):
    """Time a block of code."""
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        logger.info(f"{name} completed", duration_seconds=round(elapsed, 3))

# Usage
with timed_block("data_processing"):
    process_large_dataset()

@asynccontextmanager
async def database_transaction(conn: AsyncConnection):
    """Manage database transaction."""
    await conn.execute("BEGIN")
    try:
        yield conn
        await conn.execute("COMMIT")
    except Exception:
        await conn.execute("ROLLBACK")
        raise

# Usage
async with database_transaction(conn) as tx:
    await tx.execute("INSERT INTO users ...")
    await tx.execute("INSERT INTO audit_log ...")

パターン 4：リソースの無条件解放

例外の有無にかかわらず、__exit__ で必ずリソースをクリーンアップします。

class FileProcessor:
    """Process file with guaranteed cleanup."""

    def __init__(self, path: str) -> None:
        self._path = path
        self._file: IO | None = None
        self._temp_files: list[Path] = []

    def __enter__(self) -> "FileProcessor":
        self._file = open(self._path, "r")
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Clean up all resources unconditionally."""
        # Close main file
        if self._file is not None:
            self._file.close()

        # Clean up any temporary files
        for temp_file in self._temp_files:
            try:
                temp_file.unlink()
            except OSError:
                pass  # Best effort cleanup

        # Return None/False to propagate any exception

高度なパターン

パターン 5：例外を選択的に抑制する

ドキュメントに記載された特定の例外のみを抑制します。

class StreamWriter:
    """Writer that handles broken pipe gracefully."""

    def __init__(self, stream) -> None:
        self._stream = stream

    def __enter__(self) -> "StreamWriter":
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> bool:
        """Clean up, suppressing BrokenPipeError on shutdown."""
        self._stream.close()

        # Suppress BrokenPipeError (client disconnected)
        # This is expected behavior, not an error
        if exc_type is BrokenPipeError:
            return True  # Exception suppressed

        return False  # Propagate all other exceptions

パターン 6：状態を蓄積しながらストリーミングする

ストリーミング中に、逐次的なチャンクと蓄積した状態の両方を維持します。

from collections.abc import Generator
from dataclasses import dataclass, field

@dataclass
class StreamingResult:
    """Accumulated streaming result."""

    chunks: list[str] = field(default_factory=list)
    _finalized: bool = False

    @property
    def content(self) -> str:
        """Get accumulated content."""
        return "".join(self.chunks)

    def add_chunk(self, chunk: str) -> None:
        """Add chunk to accumulator."""
        if self._finalized:
            raise RuntimeError("Cannot add to finalized result")
        self.chunks.append(chunk)

    def finalize(self) -> str:
        """Mark stream complete and return content."""
        self._finalized = True
        return self.content

def stream_with_accumulation(
    response: StreamingResponse,
) -> Generator[tuple[str, str], None, str]:
    """Stream response while accumulating content.

    Yields:
        Tuple of (accumulated_content, new_chunk) for each chunk.

    Returns:
        Final accumulated content.
    """
    result = StreamingResult()

    for chunk in response.iter_content():
        result.add_chunk(chunk)
        yield result.content, chunk

    return result.finalize()

パターン 7：文字列を効率よく蓄積する

蓄積時に O(n²) の文字列連結を避けます。

def accumulate_stream(stream) -> str:
    """Efficiently accumulate stream content."""
    # BAD: O(n²) due to string immutability
    # content = ""
    # for chunk in stream:
    #     content += chunk  # Creates new string each time

    # GOOD: O(n) with list and join
    chunks: list[str] = []
    for chunk in stream:
        chunks.append(chunk)
    return "".join(chunks)  # Single allocation

パターン 8：ストリームのメトリクスを追跡する

最初のバイトが届くまでの時間と、ストリーミング全体にかかる時間を測定します。

import time
from collections.abc import Generator

def stream_with_metrics(
    response: StreamingResponse,
) -> Generator[str, None, dict]:
    """Stream response while collecting metrics.

    Yields:
        Content chunks.

    Returns:
        Metrics dictionary.
    """
    start = time.perf_counter()
    first_chunk_time: float | None = None
    chunk_count = 0
    total_bytes = 0

    for chunk in response.iter_content():
        if first_chunk_time is None:
            first_chunk_time = time.perf_counter() - start

        chunk_count += 1
        total_bytes += len(chunk.encode())
        yield chunk

    total_time = time.perf_counter() - start

    return {
        "time_to_first_byte_ms": round((first_chunk_time or 0) * 1000, 2),
        "total_time_ms": round(total_time * 1000, 2),
        "chunk_count": chunk_count,
        "total_bytes": total_bytes,
    }

パターン 9：ExitStack で複数のリソースを管理する

数が動的に変化するリソースをすっきりと処理します。

from contextlib import ExitStack, AsyncExitStack
from pathlib import Path

def process_files(paths: list[Path]) -> list[str]:
    """Process multiple files with automatic cleanup."""
    results = []

    with ExitStack() as stack:
        # Open all files - they'll all be closed when block exits
        files = [stack.enter_context(open(p)) for p in paths]

        for f in files:
            results.append(f.read())

    return results

async def process_connections(hosts: list[str]) -> list[dict]:
    """Process multiple async connections."""
    results = []

    async with AsyncExitStack() as stack:
        connections = [
            await stack.enter_async_context(connect_to_host(host))
            for host in hosts
        ]

        for conn in connections:
            results.append(await conn.fetch_data())

    return results

ベストプラクティスのまとめ

1. 必ずコンテキストマネージャーを使う — クリーンアップが必要なあらゆるリソースに使用します
2. 必ずクリーンアップする — 例外が発生しても __exit__ は実行されます
3. 意図せず例外を抑制しない — 抑制する意図がない限り False を返します
4. @contextmanager を使う — 単純なリソースパターンに適しています
5. 両方のプロトコルを実装する — with と手動での管理の両方をサポートします
6. ExitStack を使う — 数が動的に変化するリソースに適しています
7. 効率よく蓄積する — 文字列連結ではなく、リストと join を使います
8. メトリクスを追跡する — ストリーミングでは、最初のバイトが届くまでの時間が重要です
9. 動作をドキュメントに記載する — 特に例外の抑制について記載します
10. クリーンアップの経路をテストする — エラー発生時にリソースが解放されることを確認します

---

注意点

• ジェネレーター内の with open()：ファイルが閉じられるのは反復の終了時ではなく、ジェネレーターが GC されたときです — CPython（参照カウント方式）では問題ありませんが、PyPy／IronPython ではリスクがあります。
• ExitStack：enter_context は __enter__／__exit__ プロトコル用、push はクリーンアップのみを行うコールバック用です — 順序を誤って混在させると、クリーンアップチェーンで例外が握りつぶされる可能性があります。
• 非同期コンテキストマネージャーでは必ず async with を使ってください — 非同期マネージャーに with を使うと、コルーチンがそのまま返され、クリーンアップは実行されません。
• Windows では tempfile.NamedTemporaryFile を開いたまま再度開くことはできません — クロスプラットフォームで使用するには、delete=False と明示的なクリーンアップを使います。
• contextlib.suppress(Exception) は、BaseException を渡すと KeyboardInterrupt も含めて抑制します — 抑制する例外は具体的に指定してください。