Dicas para escrever scripts Python mais robustos

1. Tratamento de Exceções e Validação de Entradas

Um dos pilares para scripts robustos é o tratamento adequado de exceções. Evite o uso de except genérico, que pode mascarar erros inesperados e dificultar a depuração.

import argparse
import sys

def processar_dados(arquivo_entrada):
    """Processa dados do arquivo de entrada."""
    try:
        with open(arquivo_entrada, 'r') as f:
            dados = f.read()
    except FileNotFoundError:
        print(f"Erro: Arquivo '{arquivo_entrada}' não encontrado.")
        sys.exit(1)
    except PermissionError:
        print(f"Erro: Sem permissão para ler '{arquivo_entrada}'.")
        sys.exit(1)
    except Exception as e:
        print(f"Erro inesperado ao ler arquivo: {e}")
        sys.exit(1)
    return dados

def validar_numeros(valor):
    """Valida se o valor é um número positivo."""
    assert isinstance(valor, (int, float)), "Valor deve ser numérico"
    assert valor > 0, "Valor deve ser positivo"
    return valor

def main():
    parser = argparse.ArgumentParser(description='Processador de dados robusto')
    parser.add_argument('arquivo', help='Caminho do arquivo de entrada')
    parser.add_argument('--limite', type=float, default=100.0,
                       help='Limite de processamento (padrão: 100.0)')
    args = parser.parse_args()

    dados = processar_dados(args.arquivo)
    limite_valido = validar_numeros(args.limite)
    print(f"Dados carregados. Limite: {limite_valido}")

if __name__ == "__main__":
    main()

2. Gerenciamento de Recursos e Contextos

Utilize gerenciadores de contexto para garantir o fechamento adequado de recursos, mesmo em cenários de erro.

from contextlib import contextmanager
import sqlite3
import psutil

@contextmanager
def gerenciar_conexao_banco(caminho_bd):
    """Gerenciador de contexto para conexão com banco de dados."""
    conn = None
    try:
        conn = sqlite3.connect(caminho_bd)
        yield conn
    finally:
        if conn:
            conn.close()

def verificar_recursos_sistema():
    """Verifica limites de recursos do sistema."""
    memoria = psutil.virtual_memory()
    if memoria.percent > 90:
        raise MemoryError("Memória do sistema acima de 90%")

    with open('/proc/sys/fs/file-max', 'r') as f:
        limite_arquivos = int(f.read().strip())

    arquivos_atuais = len(psutil.Process().open_files())
    if arquivos_atuais > limite_arquivos * 0.8:
        print("Aviso: Atingindo limite de descritores de arquivo")

# Exemplo de uso
with gerenciar_conexao_banco('dados.db') as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM tabela")
    resultados = cursor.fetchall()

3. Logging e Monitoramento Robusto

Substitua print por logging estruturado para melhor rastreabilidade e monitoramento.

import logging
import uuid
from logging.handlers import RotatingFileHandler

def configurar_logging():
    """Configura sistema de logging com rotação de arquivos."""
    identificador_execucao = str(uuid.uuid4())[:8]

    logger = logging.getLogger('script_robusto')
    logger.setLevel(logging.DEBUG)

    # Handler para arquivo com rotação
    file_handler = RotatingFileHandler(
        'logs/script.log',
        maxBytes=10485760,  # 10MB
        backupCount=5
    )
    file_handler.setLevel(logging.INFO)

    # Handler para console
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)

    # Formato com identificador único
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - '
        f'[{identificador_execucao}] - %(message)s'
    )
    file_handler.setFormatter(formatter)
    console_handler.setFormatter(formatter)

    logger.addHandler(file_handler)
    logger.addHandler(console_handler)

    return logger

logger = configurar_logging()

def processar_lote(dados):
    """Processa um lote de dados com logging detalhado."""
    logger.info(f"Iniciando processamento de {len(dados)} registros")
    try:
        for i, item in enumerate(dados):
            logger.debug(f"Processando item {i+1}: {item}")
            # Lógica de processamento
        logger.info("Processamento concluído com sucesso")
    except Exception as e:
        logger.error(f"Falha no processamento: {e}", exc_info=True)
        raise

4. Estruturação Modular e Reutilização

Organize seu código em módulos coesos com ponto de entrada claro e configurações externas.

# config.yaml
# database:
#   host: localhost
#   port: 5432
# logging:
#   level: INFO
#   file: logs/app.log

import yaml
import os
from dataclasses import dataclass

@dataclass
class Configuracao:
    database_host: str
    database_port: int
    log_level: str
    log_file: str

def carregar_configuracao(caminho_config=None):
    """Carrega configuração de arquivo YAML ou variáveis de ambiente."""
    config_padrao = {
        'database': {'host': 'localhost', 'port': 5432},
        'logging': {'level': 'INFO', 'file': 'logs/app.log'}
    }

    if caminho_config and os.path.exists(caminho_config):
        with open(caminho_config, 'r') as f:
            config = yaml.safe_load(f)
    else:
        config = config_padrao

    # Sobrescrever com variáveis de ambiente
    config['database']['host'] = os.getenv('DB_HOST', config['database']['host'])
    config['database']['port'] = int(os.getenv('DB_PORT', config['database']['port']))

    return Configuracao(
        database_host=config['database']['host'],
        database_port=config['database']['port'],
        log_level=config['logging']['level'],
        log_file=config['logging']['file']
    )

# Módulo separado para funções de processamento
# processamento.py
def calcular_media(valores):
    """Calcula a média de uma lista de valores."""
    if not valores:
        return 0
    return sum(valores) / len(valores)

5. Tratamento de Erros e Graceful Degradation

Implemente retry com backoff para operações externas e fallbacks seguros.

import time
import random
from functools import wraps

def retry(max_tentativas=3, delay_base=1, backoff_factor=2):
    """Decorator para retry com backoff exponencial."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            ultimo_erro = None
            for tentativa in range(max_tentativas):
                try:
                    return func(*args, **kwargs)
                except (ConnectionError, TimeoutError) as e:
                    ultimo_erro = e
                    if tentativa < max_tentativas - 1:
                        delay = delay_base * (backoff_factor ** tentativa)
                        delay += random.uniform(0, 0.1 * delay)  # Jitter
                        print(f"Tentativa {tentativa + 1} falhou. "
                              f"Tentando novamente em {delay:.2f}s...")
                        time.sleep(delay)
            raise ultimo_erro
        return wrapper
    return decorator

def fallback_seguro(valor_padrao=None):
    """Decorator para fallback em caso de falha."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                print(f"Falha na operação: {e}. Usando valor padrão.")
                return valor_padrao
        return wrapper
    return decorator

@retry(max_tentativas=3)
@fallback_seguro(valor_padrao=[])
def buscar_dados_api(url):
    """Busca dados de uma API externa com retry e fallback."""
    import requests
    resposta = requests.get(url, timeout=5)
    resposta.raise_for_status()
    return resposta.json()

# Notificação automática
def notificar_falha(mensagem):
    """Envia notificação em caso de falha crítica."""
    import smtplib
    from email.mime.text import MIMEText

    msg = MIMEText(f"Falha crítica no script:\n{mensagem}")
    msg['Subject'] = 'Alerta: Script Python - Falha Crítica'
    msg['From'] = 'monitor@exemplo.com'
    msg['To'] = 'admin@exemplo.com'

    with smtplib.SMTP('localhost') as servidor:
        servidor.send_message(msg)

6. Performance e Otimização Consciente

Utilize profiling e estruturas de dados adequadas para evitar gargalos.

import cProfile
import pstats
from timeit import timeit

def analisar_performance():
    """Analisa performance do código com cProfile."""
    profiler = cProfile.Profile()
    profiler.enable()

    # Código a ser analisado
    dados = list(range(1000000))
    resultado = sum(dados)

    profiler.disable()
    stats = pstats.Stats(profiler)
    stats.sort_stats('cumulative')
    stats.print_stats(10)

# Uso de estruturas de dados adequadas
def buscar_rapido(lista_alvos):
    """Usa set para busca O(1) em vez de lista O(n)."""
    alvos_set = set(lista_alvos)
    dados = list(range(100000))

    for item in dados:
        if item in alvos_set:  # Busca eficiente
            yield item

# Generators para economia de memória
def gerar_dados_grande_volume():
    """Gera dados em lotes usando generator."""
    for i in range(1000000):
        yield i ** 2

# Paralelismo seguro
from concurrent.futures import ThreadPoolExecutor, as_completed

def processar_paralelo(dados):
    """Processa dados em paralelo com controle de threads."""
    with ThreadPoolExecutor(max_workers=4) as executor:
        futuros = {executor.submit(processar_item, item): item 
                  for item in dados}

        for futuro in as_completed(futuros):
            try:
                resultado = futuro.result()
                yield resultado
            except Exception as e:
                print(f"Erro ao processar item: {e}")

7. Testabilidade e Manutenção

Escreva testes unitários e utilize type hints para melhor documentação.

from typing import List, Optional, Dict, Any
import pytest
from unittest.mock import Mock, patch

def calcular_media(valores: List[float]) -> float:
    """Calcula a média de uma lista de números.

    Args:
        valores: Lista de números para calcular a média

    Returns:
        Média dos valores

    Raises:
        ValueError: Se a lista estiver vazia
    """
    if not valores:
        raise ValueError("Lista de valores não pode estar vazia")
    return sum(valores) / len(valores)

class ProcessadorDados:
    """Classe para processamento de dados com documentação."""

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.config = config or {}
        self.logger = configurar_logging()

    def processar(self, dados: List[float]) -> Dict[str, Any]:
        """Processa uma lista de dados e retorna estatísticas.

        Args:
            dados: Lista de números para processar

        Returns:
            Dicionário com estatísticas calculadas
        """
        return {
            'media': calcular_media(dados),
            'maximo': max(dados),
            'minimo': min(dados),
            'total': len(dados)
        }

# Testes unitários com pytest
def test_calcular_media():
    """Testa a função calcular_media."""
    assert calcular_media([1, 2, 3]) == 2.0
    assert calcular_media([0, 0, 0]) == 0.0
    assert calcular_media([1.5, 2.5]) == 2.0

    with pytest.raises(ValueError):
        calcular_media([])

def test_processador_com_mock():
    """Testa ProcessadorDados com mock de dependências."""
    with patch('modulo.configurar_logging') as mock_log:
        mock_log.return_value = Mock()
        processador = ProcessadorDados()

        resultado = processador.processar([1, 2, 3])
        assert resultado['media'] == 2.0
        assert resultado['total'] == 3

Referências