python-resource-management

出典の明記：本スキルの内容は wshobson/agents に由来します（MIT License, Copyright (c) 2024 Seth Hobson。ライセンス全文は同じディレクトリの LICENSE.txt をご覧ください）。 使用、変更、再配布には当該 MIT ライセンスの条項に従い、著作権表示を保持する必要があります。

日本語での説明

本スキルは、フルスタック開発に関するベストプラクティスとパターンを提供します。

---

name: python-resource-management description: コンテキストマネージャー、クリーンアップのパターン、ストリーミングを使用した Python のリソース管理です。接続やファイルハンドルの管理、クリーンアップロジックの実装、状態を蓄積するストリーミングレスポンスの構築に使用します。

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

詳細な実例とパターン

詳細なセクション（## Advanced Patterns から始まります）は references/details.md にあります。上記の概要だけでは不十分な場合は、そのファイルを参照してください。

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

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

対応範囲

✅ 得意なこと

• フルスタック開発に関するベストプラクティス
• コードアーキテクチャ設計のガイダンス
• パフォーマンス最適化の提案

⚠️ 情報が必要なこと

• 具体的なプロジェクトのコンテキスト
• 技術スタックの情報

❌ 対応範囲外

• 技術分野以外のビジネス上の意思決定
• 具体的なコード実装（ガイダンスのみを提供）

ワークフロー

Step 1: 要件分析

プロジェクトの要件と技術的な制約を理解します

Step 2: アーキテクチャ設計

システムアーキテクチャとコンポーネントの分割を設計します

Step 3: 実装ガイダンス

具体的な実装の提案とコードパターンを提示します

Step 4: コードレビュー

コードの品質とベストプラクティスへの準拠状況を確認します

Step 5: 最適化の提案

パフォーマンスの最適化と改善案を提示します

注意点

1. 過剰設計

単純な場面で複雑なデザインパターンを使用しないようにします

2. パフォーマンスの軽視

特に本番環境では、パフォーマンスの最適化を軽視しないでください

3. セキュリティへの配慮

セキュリティとデータ保護を常に考慮してください

4. テストカバレッジ

特に境界条件について、十分なテストカバレッジを確保してください

5. ドキュメントの保守

ドキュメントをコードと同期して更新してください