Skill v1.0.1

currentLLM-judged scan95/100

wshobson/agents/python-resource-management

4 files

──Details

PublishedJune 20, 2026 at 08:07 PM

Content Hashsha256:1d60cbc5ce7a9e86...

Git SHAcc37bfdd292c

Bump Typepatch

Compare with v1.0.0

──Files

Files (1 file, 6.9 KB)

SKILL.md6.9 KBactive

SKILL.md · 245 lines · 6.9 KB

version: "1.0.1" name: python-resource-management description: 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.

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
@contextmanager
def 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 needed
db = 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)
# Usage
async 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, 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 ...")

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

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

← v1.0.0 All versions