Python Resource Management

作者 wshobson46891e7e60da無授權條款收錄於 2026年10月8日更新於 2026年10月8日

Python resource management with context managers, cleanup patterns, and streaming. Use when managing connections, file handles, implementing cleanup logic, or building streaming responses with accumulated state.

AI 產生的概覽

指導 Python 開發者使用情境管理器、清理模式與串流處理來確定性地管理資源。

功能
此技能提供 Python 資源管理的指引,涵蓋情境管理器、enter/exit 與非同步協定、無條件清理、例外處理,以及帶有累積狀態的串流處理。它提出以類別與裝飾器為基礎的模式,包括資料庫連線、連線池、檔案控制代碼與交易,並附上程式碼範例。它也包含最佳實務摘要,並指向參考檔案以了解進階模式。
適用情境
適用於撰寫或審查需要可靠釋放資料庫連線、檔案控制代碼、網路通訊端等資源的 Python 程式碼時。也適用於實作自訂或非同步情境管理器、巢狀清理,或帶有累積狀態的串流回應時。
執行需求
不包含指令碼,僅為說明性內容。範例涉及 psycopg、asyncpg、structlog 與 contextlib 等 Python 函式庫,但技能本身除了供代理閱讀外無需其他條件。

Python Resource Management

Manage resources deterministically using context managers. Resources like database connections, file handles, and network sockets should be released reliably, even when exceptions occur.

When to Use This Skill

  • Managing database connections and connection pools
  • Working with file handles and I/O
  • Implementing custom context managers
  • Building streaming responses with state
  • Handling nested resource cleanup
  • Creating async context managers

Core Concepts

1. Context Managers

The with statement ensures resources are released automatically, even on exceptions.

2. Protocol Methods

__enter__/__exit__ for sync, __aenter__/__aexit__ for async resource management.

3. Unconditional Cleanup

__exit__ always runs, regardless of whether an exception occurred.

4. Exception Handling

Return True from __exit__ to suppress exceptions, False to propagate them.

Quick Start

python
from contextlib import contextmanager
@contextmanagerdef managed_resource():    resource = acquire_resource()    try:        yield resource    finally:        resource.cleanup()
with managed_resource() as r:    r.do_work()

Fundamental Patterns

Pattern 1: Class-Based Context Manager

Implement the context manager protocol for complex resources.

python
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 neededdb = DatabaseConnection(dsn)db.connect()try:    result = db.execute(query)finally:    db.close()

Pattern 2: Async Context Manager

For async resources, implement the async protocol.

python
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)
# Usageasync with AsyncDatabasePool(dsn) as pool:    users = await pool.execute("SELECT * FROM users WHERE active = $1", True)

Pattern 3: Using @contextmanager Decorator

Simplify context managers with the decorator for straightforward cases.

python
from contextlib import contextmanager, asynccontextmanagerimport timeimport structlog
logger = structlog.get_logger()
@contextmanagerdef 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))
# Usagewith timed_block("data_processing"):    process_large_dataset()
@asynccontextmanagerasync 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
# Usageasync with database_transaction(conn) as tx:    await tx.execute("INSERT INTO users ...")    await tx.execute("INSERT INTO audit_log ...")

Pattern 4: Unconditional Resource Release

Always clean up resources in __exit__, regardless of exceptions.

python
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

Detailed worked examples and patterns

Detailed sections (starting with ## Advanced Patterns) live in references/details.md. Read that file when the navigation summary above is insufficient.

Best Practices Summary

  1. Always use context managers - For any resource that needs cleanup
  2. Clean up unconditionally - __exit__ runs even on exception
  3. Don't suppress unexpectedly - Return False unless suppression is intentional
  4. Use @contextmanager - For simple resource patterns
  5. Implement both protocols - Support with and manual management
  6. Use ExitStack - For dynamic numbers of resources
  7. Accumulate efficiently - List + join, not string concatenation
  8. Track metrics - Time-to-first-byte matters for streaming
  9. Document behavior - Especially exception suppression
  10. Test cleanup paths - Verify resources are released on errors

來源與署名

來源:wshobson/agents位於plugins/python-development/skills/python-resource-management提交46891e7

授權條款: 無授權條款

內容歸原作者所有。SourceWeft 從公開儲存庫中收錄這些內容。

檢舉或申請下架