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

🎯 Demonstração Prática: Pipeline Funcional de Processamento de Logs em Tempo Real

O exemplo abaixo implementa um pipeline analítico em streaming construído sob o paradigma funcional:

  • Encadeamento de geradores (yield) operando com consumo estritamente constante de memória ($O(1)$ RAM).
  • Utilização de @functools.lru_cache para memoização de resolução de geolocalização por IP.
  • Aplicação de functools.reduce e functools.partial para redução estatística agregada de tráfego.
  • Filtros declarativos e transformações lazy com Type Hints estritos.
"""
demonstracao_funcional_e_geradores.py
Pipeline de Streaming Lazy com Geradores, functools (lru_cache, reduce) e Lambdas.
Compatível com Python 3.12+.
"""

from __future__ import annotations

from collections.abc import Generator, Iterable
from dataclasses import dataclass
from decimal import Decimal
from functools import lru_cache, partial, reduce
import re
import time
from typing import Final

# ==============================================================================
# 1. MODELO DE LOG E ENRIQUECIMENTO GEOGRÁFICO MEMOIZADO
# ==============================================================================
@dataclass(slots=True)
class EventoLogAcesso:
    timestamp: str
    ip_origem: str
    pais_origem: str
    metodo_http: str
    url_endpoint: str
    status_code: int
    latencia_ms: float
    bytes_enviados: int

# Memoização automática com LRU Cache: Evita consultas repetidas de IP
@lru_cache(maxsize=1024)
def resolver_pais_por_ip(ip: str) -> str:
    """Simula resolução de ASN / GeoIP com cache em memória."""
    if ip.startswith("200.") or ip.startswith("189."):
        return "Brasil"
    elif ip.startswith("104.") or ip.startswith("54."):
        return "Estados Unidos"
    elif ip.startswith("140."):
        return "Japão"
    return "Desconhecido"

# ==============================================================================
# 2. ESTÁGIOS DO PIPELINE DE STREAMING (GERADORES COM YIELD)
# ==============================================================================
def estagio_1_gerar_logs_brutos(linhas_brutas: Iterable[str]) -> Generator[str, None, None]:
    """Lê linhas brutas ignorando linhas em branco ou comentários."""
    for linha in linhas_brutas:
        linha_limpa = linha.strip()
        if linha_limpa and not linha_limpa.startswith("#"):
            yield linha_limpa

def estagio_2_parsear_eventos(linhas_stream: Generator[str, None, None]) -> Generator[EventoLogAcesso, None, None]:
    """Converte strings em objetos de domínio tipados."""
    padrao_regex = re.compile(
        r"^(?P<ts>[\d\-:T]+)\s+(?P<ip>[\d\.]+)\s+(?P<metodo>[A-Z]+)\s+(?P<url>[^\s]+)\s+(?P<status>\d{3})\s+(?P<lat>[\d\.]+)\s+(?P<bytes>\d+)$"
    )
    for linha in linhas_stream:
        match = padrao_regex.match(linha)
        if match:
            d = match.groupdict()
            ip = d["ip"]
            yield EventoLogAcesso(
                timestamp=d["ts"],
                ip_origem=ip,
                pais_origem=resolver_pais_por_ip(ip), # Chamada cacheada
                metodo_http=d["metodo"],
                url_endpoint=d["url"],
                status_code=int(d["status"]),
                latencia_ms=float(d["lat"]),
                bytes_enviados=int(d["bytes"])
            )

def estagio_3_filtrar_anomalias(
    eventos: Generator[EventoLogAcesso, None, None],
    limiar_latencia: float = 200.0
) -> Generator[EventoLogAcesso, None, None]:
    """Filtra requisições lentas ou erros de servidor (HTTP 5xx)."""
    for ev in eventos:
        if ev.latencia_ms >= limiar_latencia or ev.status_code >= 500:
            yield ev

# ==============================================================================
# 3. REDUÇÃO FUNCIONAL COM FUNCTOOLS.REDUCE
# ==============================================================================
@dataclass(slots=True)
class EstatisticasTrafego:
    total_requisicoes: int = 0
    total_bytes: int = 0
    latencia_acumulada: float = 0.0

def acumulador_metricas(acum: EstatisticasTrafego, ev: EventoLogAcesso) -> EstatisticasTrafego:
    """Função pura redutora para agregação estatística."""
    acum.total_requisicoes += 1
    acum.total_bytes += ev.bytes_enviados
    acum.latencia_acumulada += ev.latencia_ms
    return acum

# ==============================================================================
# 4. EXECUÇÃO DO PIPELINE
# ==============================================================================
if __name__ == "__main__":
    logs_simulados = [
        "2026-09-28T10:00:01 200.147.3.15 GET /api/v1/produtos 200 45.2 1250",
        "2026-09-28T10:00:02 54.210.88.9 POST /api/v1/checkout 500 450.8 450", # Anomalia: 500 e latência
        "2026-09-28T10:00:03 200.147.3.15 GET /api/v1/produtos 200 32.1 1250", # Hit de Cache no GeoIP
        "2026-09-28T10:00:04 140.22.90.11 GET /api/v1/status 200 12.0 120",
        "2026-09-28T10:00:05 189.40.12.88 POST /api/v1/pagamentos 200 310.5 890", # Anomalia: Latência alta
        "2026-09-28T10:00:06 54.210.88.9 GET /api/v1/pedidos 200 55.4 3400",
    ]

    print("=" * 80)
    print("      PIPELINE FUNCIONAL EM STREAMING LAZY COM GERADORES E FUNCTOOLS")
    print("=" * 80)

    # Montagem declarativa do pipeline de streaming
    g_bruto = estagio_1_gerar_logs_brutos(logs_simulados)
    g_parsed = estagio_2_parsear_eventos(g_bruto)
    g_anomalias = estagio_3_filtrar_anomalias(g_parsed, limiar_latencia=200.0)

    print(">> Incidentes Críticos Filtrados no Fluxo:")
    print(f"{'Horário':<20} | {'Origem':<14} | {'País':<14} | {'Status':<7} | {'Latência':<10}")
    print("-" * 80)

    lista_anomalias: list[EventoLogAcesso] = []
    for alerta in g_anomalias:
        lista_anomalias.append(alerta)
        print(
            f"{alerta.timestamp:<20} | "
            f"{alerta.ip_origem:<14} | "
            f"{alerta.pais_origem:<14} | "
            f"HTTP {alerta.status_code:<2} | "
            f"{alerta.latencia_ms:>6.1f} ms"
        )

    # Redução funcional dos incidentes com reduce
    metricas = reduce(acumulador_metricas, lista_anomalias, EstatisticasTrafego())

    print("-" * 80)
    print(f"Total de Anomalias Registradas : {metricas.total_requisicoes}")
    print(f"Volume Trafegado em Falhas     : {metricas.total_bytes:,} bytes")
    if metricas.total_requisicoes > 0:
        media_lat = metricas.latencia_acumulada / metricas.total_requisicoes
        print(f"Latência Média dos Incidentes  : {media_lat:.2f} ms")

    print("-" * 80)
    print(f"Telemetria do Cache LRU GeoIP : {resolver_pais_por_ip.cache_info()}")
    print("=" * 80)