Especialização em Sistemas com Python 3.12+ • Capítulo 012

🎯 Demonstração Prática: Pipeline de ETL de Arquivos com Escrita Atômica e Context Managers

O exemplo abaixo implementa um pipeline de extração, transformação e carga (ETL) no sistema de arquivos:

  • Navegação orientada a objetos com pathlib.Path.
  • Gerenciador de contexto customizado com @contextmanager para telemetria de I/O.
  • Leitura e parsing de dados brutos tabulares em CSV via csv.DictReader(..., delimiter=";").
  • Escrita atômica resiliente com arquivo temporário e Path.replace(), prevenindo corrupção em disco.
  • Exportação final em JSON com indentação padronizada e codificação UTF-8 explícita.
"""
demonstracao_arquivos_e_contexto.py
Pipeline de Processamento CSV/JSON com pathlib.Path, Context Managers e Escrita Atômica.
Compatível com Python 3.12+.
"""

from __future__ import annotations

from contextlib import contextmanager
import csv
from decimal import Decimal, ROUND_HALF_EVEN
import json
import os
from pathlib import Path
import time
from typing import Any, Iterator

# ==============================================================================
# 1. GERENCIADOR DE CONTEXTO CUSTOMIZADO PARA TELEMETRIA DE I/O
# ==============================================================================
@contextmanager
def monitorar_io(nome_etapa: str) -> Iterator[None]:
    """Mede e registra o tempo de processamento em operações de disco."""
    t_inicio = time.perf_counter()
    print(f"  [I/O INÍCIO] {nome_etapa}...")
    try:
        yield
    finally:
        t_fim = time.perf_counter()
        duracao_ms = (t_fim - t_inicio) * 1000
        print(f"  [I/O CONCLUÍDO] {nome_etapa} finalizado em {duracao_ms:.3f} ms.")

# ==============================================================================
# 2. GRAVAÇÃO ATÔMICA SEGURA DE ARQUIVOS
# ==============================================================================
def gravar_json_atomicamente(caminho_final: Path, dados: dict[str, Any]) -> None:
    """
    Grava os dados em arquivo temporário com flush físico e renomeia atomicamente.
    Elimina o risco de corrupção caso o processo seja interrompido abruptamente.
    """
    caminho_final.parent.mkdir(parents=True, exist_ok=True)
    caminho_temp = caminho_final.with_suffix(".tmp")

    with caminho_temp.open(mode="w", encoding="utf-8") as f:
        json.dump(dados, f, indent=2, ensure_ascii=False)
        f.flush()
        os.fsync(f.fileno()) # Força a gravação física no hardware do disco

    # Operação atômica no sistema de arquivos
    caminho_temp.replace(caminho_final)

# ==============================================================================
# 3. PIPELINE DE ETL (CSV -> TRANSFORMAÇÃO -> JSON)
# ==============================================================================
def executar_pipeline_vendas(caminho_csv: Path, caminho_json: Path) -> dict[str, Any]:
    relatorio_consolidado: dict[str, Any] = {
        "arquivo_origem": caminho_csv.name,
        "total_registros": 0,
        "faturamento_por_departamento": {},
        "valor_total_faturado": "0.00"
    }

    faturamento_dept: dict[str, Decimal] = {}
    faturamento_global = Decimal("0.00")

    with monitorar_io(f"Leitura e parsing do CSV '{caminho_csv.name}'"):
        with caminho_csv.open(mode="r", encoding="utf-8") as arquivo_entrada:
            leitor = csv.DictReader(arquivo_entrada, delimiter=";")
            for linha in leitor:
                departamento = linha["departamento"].strip()
                valor = Decimal(linha["valor"].replace(",", "."))

                faturamento_dept[departamento] = faturamento_dept.get(departamento, Decimal("0.00")) + valor
                faturamento_global += valor
                relatorio_consolidado["total_registros"] += 1

    # Formatação contábil para saída
    relatorio_consolidado["faturamento_por_departamento"] = {
        dept: f"R$ {val.quantize(Decimal('0.01'), ROUND_HALF_EVEN):,.2f}"
        for dept, val in faturamento_dept.items()
    }
    relatorio_consolidado["valor_total_faturado"] = f"R$ {faturamento_global.quantize(Decimal('0.01'), ROUND_HALF_EVEN):,.2f}"

    with monitorar_io(f"Gravação atômica do consolidado JSON '{caminho_json.name}'"):
        gravar_json_atomicamente(caminho_json, relatorio_consolidado)

    return relatorio_consolidado

# ==============================================================================
# 4. EXECUÇÃO DEMONSTRATIVA
# ==============================================================================
if __name__ == "__main__":
    diretorio_trabalho = Path("temp_pipeline_demo")
    arquivo_csv_teste = diretorio_trabalho / "transacoes_filial.csv"
    arquivo_json_saida = diretorio_trabalho / "auditoria_consolidada.json"

    print("=" * 80)
    print("      PIPELINE DE ARQUIVOS ROBUSTO COM PATHLIB E GRAVAÇÃO ATÔMICA")
    print("=" * 80)

    # 1. Criação do arquivo de teste com CSV em UTF-8
    diretorio_trabalho.mkdir(parents=True, exist_ok=True)
    conteudo_csv_bruto = (
        "id_transacao;departamento;valor;data\n"
        "TX-101;Eletrônicos;4500.50;2026-09-28\n"
        "TX-102;Mobiliário Corporativo;12800.00;2026-09-28\n"
        "TX-103;Eletrônicos;3200.00;2026-09-28\n"
        "TX-104;Suprimentos de TI;890.90;2026-09-28\n"
        "TX-105;Mobiliário Corporativo;1500.00;2026-09-28\n"
    )
    arquivo_csv_teste.write_text(conteudo_csv_bruto, encoding="utf-8")
    print(f">> Arquivo de entrada simulado gerado em: {arquivo_csv_teste}")
    print("-" * 80)

    # 2. Execução do pipeline
    resultado = executar_pipeline_vendas(arquivo_csv_teste, arquivo_json_saida)

    print("-" * 80)
    print("Resultado da Consolidação Contábil:")
    print(f"Total de Linhas Processadas : {resultado['total_registros']}")
    print(f"Faturamento Global Apurado  : {resultado['valor_total_faturado']}")
    print("\nFaturamento por Departamento:")
    for dept, subtotal in resultado["faturamento_por_departamento"].items():
        print(f"  * {dept:<26} : {subtotal}")

    print("\n" + "-" * 80)
    print(f"Conteúdo Gravado em '{arquivo_json_saida.name}' via Escrita Atômica:")
    print(arquivo_json_saida.read_text(encoding="utf-8"))

    # Limpeza de arquivos de teste temporários
    arquivo_csv_teste.unlink()
    arquivo_json_saida.unlink()
    diretorio_trabalho.rmdir()
    print("-" * 80)
    print(">> Limpeza de ambiente concluída com sucesso.")
    print("=" * 80)