Gerenciamento de Recursos em Python

Gerencie recursos de forma determinística usando gerenciadores de contexto. Recursos como conexões de banco de dados, manipuladores de arquivos e soquetes de rede devem ser liberados de forma confiável, mesmo quando ocorrem exceções.

Quando Usar Esta Habilidade

• Gerenciamento de conexões de banco de dados e pools de conexão
• Trabalhando com manipuladores de arquivos e I/O
• Implementando gerenciadores de contexto personalizados
• Construindo respostas de streaming com estado
• Lidando com limpeza de recursos aninhados
• Criando gerenciadores de contexto assíncronos

Conceitos Principais

1. Gerenciadores de Contexto

A instrução with garante que os recursos sejam liberados automaticamente, mesmo em caso de exceções.

2. Métodos de Protocolo

__enter__/__exit__ para gerenciamento de recursos síncronos, __aenter__/__aexit__ para gerenciamento de recursos assíncronos.

3. Limpeza Incondicional

__exit__ sempre é executado, independentemente de ter ocorrido uma exceção.

4. Tratamento de Exceções

Retorne True de __exit__ para suprimir exceções, False para propagá-las.

Início Rápido

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

Padrões Fundamentais

Padrão 1: Gerenciador de Contexto Baseado em Classe

Implemente o protocolo de gerenciador de contexto para recursos complexos.

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

Padrão 2: Gerenciador de Contexto Assíncrono

Para recursos assíncronos, implemente o protocolo assíncrono.

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)

Padrão 3: Usando o Decorador @contextmanager

Simplifique gerenciadores de contexto com o decorador para casos diretos.

from contextlib import contextmanager, asynccontextmanager
import time
import structlog

logger = structlog.get_logger()

@contextmanager
def timed_block(name: str):
    """Cronometra um bloco de código."""
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        logger.info(f"{name} concluído", duration_seconds=round(elapsed, 3))

# Uso
with timed_block("processamento_de_dados"):
    process_large_dataset()

@asynccontextmanager
async def database_transaction(conn: AsyncConnection):
    """Gerencia transação de banco de dados."""

…