python-resource-management

# 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 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