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

🎯 Demonstração Prática: Inversão de Dependência e Repositório Genérico com Protocols

O exemplo abaixo implementa o princípio da Inversão de Dependência (SOLID DIP) em uma arquitetura de serviços utilizando typing.Protocol (PEP 544):

  • Contrato estrutural de notificação (CanalNotificacaoProtocol) decorado com @runtime_checkable.
  • Contrato de repositório genérico parametrizado (RepositorioProtocol[T]).
  • Adaptadores concretos (NotificadorSlack, NotificadorEmailTransacional, RepositorioEmMemoria) que cumprem o contrato sem herança explícita.
  • Caso de uso de auditoria desacoplado que aceita qualquer estrutura compatível.
"""
demonstracao_protocols_e_dip.py
Duck Typing Estático com typing.Protocol, Generics e Arquitetura Limpa.
Compatível com Python 3.12+ (PEP 544).
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal
from typing import Final, Protocol, TypeVar, runtime_checkable
import uuid

# ==============================================================================
# 1. CONTRATOS ESTRUTURAIS (PROTOCOLOS / PORTAS)
# ==============================================================================
@runtime_checkable
class CanalNotificacaoProtocol(Protocol):
    """Porta de saída para canais de mensageria e alerta."""
    def despachar(self, destinatario: str, mensagem: str, prioridade_alta: bool = False) -> bool:
        ...

T = TypeVar("T")

class RepositorioProtocol(Protocol[T]):
    """Porta genérica de persistência de dados."""
    def salvar(self, entidade: T) -> None:
        ...

    def obter_por_id(self, id_registro: str) -> T | None:
        ...

    def listar_todos(self) -> list[T]:
        ...

# ==============================================================================
# 2. MODELO DE DOMÍNIO
# ==============================================================================
@dataclass(slots=True)
class EventoSeguranca:
    id_evento: str
    origem_ip: str
    gravidade: str
    descricao: str
    timestamp_utc: datetime

# ==============================================================================
# 3. ADAPTADORES CONCRETOS (DUCK TYPING PURO - SEM HERDAR DE PROTOCOL)
# ==============================================================================
class NotificadorSlack:
    """Implementa CanalNotificacaoProtocol sem herdar de nada explicitamente."""
    def __init__(self, canal_webhook: str) -> None:
        self.canal_webhook: str = canal_webhook

    def despachar(self, destinatario: str, mensagem: str, prioridade_alta: bool = False) -> bool:
        alerta = "🚨 [URGENTE]" if prioridade_alta else "ℹ️ [INFO]"
        print(f"  [SLACK Webhook: {self.canal_webhook}] {alerta} @{destinatario} -> {mensagem}")
        return True

class NotificadorEmailTransacional:
    """Implementa CanalNotificacaoProtocol com lógica própria de SMTP simulada."""
    def __init__(self, servidor_smtp: str) -> None:
        self.servidor_smtp: str = servidor_smtp

    def despachar(self, destinatario: str, mensagem: str, prioridade_alta: bool = False) -> bool:
        flag = "[PRIORITÁRIO]" if prioridade_alta else "[REGULAR]"
        print(f"  [SMTP: {self.servidor_smtp}] {flag} Enviando e-mail para <{destinatario}>: {mensagem}")
        return True

class RepositorioMemoria(RepositorioProtocol[EventoSeguranca]):
    """Implementa RepositorioProtocol em memória para testes e alta vazão."""
    def __init__(self) -> None:
        self._armazenamento: dict[str, EventoSeguranca] = {}

    def salvar(self, entidade: EventoSeguranca) -> None:
        self._armazenamento[entidade.id_evento] = entidade

    def obter_por_id(self, id_registro: str) -> EventoSeguranca | None:
        return self._armazenamento.get(id_registro)

    def listar_todos(self) -> list[EventoSeguranca]:
        return list(self._armazenamento.values())

# ==============================================================================
# 4. CASO DE USO DE NEGÓCIO (INVERSÃO DE DEPENDÊNCIA)
# ==============================================================================
class GerenciadorIncidentesService:
    """
    O caso de uso depende estritamente das abstrações (Protocolos).
    Pode receber qualquer implementação em tempo de execução sem refatoração.
    """
    def __init__(
        self,
        repositorio: RepositorioProtocol[EventoSeguranca],
        canais_alerta: list[CanalNotificacaoProtocol]
    ) -> None:
        self.repositorio = repositorio
        self.canais_alerta = canais_alerta

    def registrar_e_alertar(self, ip_origem: str, severidade: str, detalhe: str) -> EventoSeguranca:
        novo_id = f"EVT-{uuid.uuid4().hex[:8].upper()}"
        evento = EventoSeguranca(
            id_evento=novo_id,
            origem_ip=ip_origem,
            gravidade=severidade,
            descricao=detalhe,
            timestamp_utc=datetime.now(timezone.utc)
        )

        # 1. Persistência desacoplada
        self.repositorio.salvar(evento)

        # 2. Despacho polimórfico estrutural
        is_critico = severidade.upper() in ("CRITICO", "FATAL")
        for canal in self.canais_alerta:
            # Verificação segura em runtime via @runtime_checkable
            if isinstance(canal, CanalNotificacaoProtocol):
                canal.despachar(
                    destinatario="equipe-soc",
                    mensagem=f"Incidente {evento.id_evento} detectado no IP {evento.origem_ip}: {evento.descricao}",
                    prioridade_alta=is_critico
                )

        return evento

# ==============================================================================
# 5. EXECUÇÃO DEMONSTRATIVA
# ==============================================================================
if __name__ == "__main__":
    print("=" * 80)
    print("      POLIMORFISMO ESTRUTURAL COM TYPING.PROTOCOL (SOLID DIP)")
    print("=" * 80)

    # Injeção de dependências puramente estruturais
    repo = RepositorioMemoria()
    canais = [
        NotificadorSlack(canal_webhook="https://hooks.slack.com/services/SEC/OPS"),
        NotificadorEmailTransacional(servidor_smtp="smtp.corp.seguranca.com.br")
    ]

    servico = GerenciadorIncidentesService(repositorio=repo, canais_alerta=canais)

    print(">> Registrando incidente de severidade 'CRITICO'...")
    incidente1 = servico.registrar_e_alertar(
        ip_origem="192.168.1.105",
        severidade="CRITICO",
        detalhe="Múltiplas tentativas de força bruta no endpoint SSH administrativo."
    )

    print("\n>> Registrando incidente de severidade 'INFO'...")
    incidente2 = servico.registrar_e_alertar(
        ip_origem="10.0.0.12",
        severidade="INFO",
        detalhe="Certificado TLS renovado automaticamente via Let's Encrypt."
    )

    print("\n" + "-" * 80)
    print(f"Total de Incidentes no Repositório: {len(repo.listar_todos())}")
    for item in repo.listar_todos():
        print(f"  * [{item.id_evento}] {item.gravidade:<8} | IP: {item.origem_ip:<15} | {item.descricao}")
    print("=" * 80)