Python Resource Management

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

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()

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)

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 ...")

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