💻 Exemplos: Paradigmas Declarativo, Lógico e Reativo
Módulo 10: Paradigmas & Padrões • Bloco 1 • Demonstrações Práticas
🔄 Fio Condutor - Fase 4 (Fechamento do Bloco 1): O Carrinho de Compras implementado sob as perspectivas lógica e reativa. Aqui você encontrará: (1) um motor dedutivo em lógica de predicados pura que infere elegibilidade e descontos do carrinho por unificação e backtracking; (2) um barramento reativo assíncrono com controle de contrapressão (backpressure) para processamento de eventos de checkout em tempo real.
1. Motor Lógico Dedutivo com Unificação de Termos (Prolog em Python)
Neste exemplo, implementamos um mini-motor de inferência baseado em lógica de predicados. O desenvolvedor apenas declara fatos e regras de negócio do carrinho de compras. O algoritmo de unificação e retrocesso (backtracking) descobre automaticamente quais clientes têm direito a frete grátis e descontos promocionais:
# motor_logico_carrinho.py
from typing import Dict, List, Tuple, Any, Optional
# 1. Estruturas da Lógica de Predicados
class Termo:
def __init__(self, predicado: str, *argumentos: Any):
self.predicado = predicado
self.argumentos = list(argumentos)
def __repr__(self):
args_str = ", ".join(str(a) for a in self.argumentos)
return f"{self.predicado}({args_str})"
class Variavel:
def __init__(self, nome: str):
self.nome = nome
def __repr__(self):
return f"?{self.nome}"
class Regra:
def __init__(self, cabecalho: Termo, premissas: List[Termo]):
self.cabecalho = cabecalho # Conclusão (H)
self.premissas = premissas # Condições (B1, B2, ...)
# 2. Algoritmo de Unificação de Robinson
def unificar(x: Any, y: Any, subs: Dict[str, Any]) -> Optional[Dict[str, Any]]:
if subs is None:
return None
elif x == y:
return subs
elif isinstance(x, Variavel):
return unificar_variavel(x, y, subs)
elif isinstance(y, Variavel):
return unificar_variavel(y, x, subs)
elif isinstance(x, Termo) and isinstance(y, Termo):
if x.predicado != y.predicado or len(x.argumentos) != len(y.argumentos):
return None
novo_subs = dict(subs)
for ax, ay in zip(x.argumentos, y.argumentos):
novo_subs = unificar(ax, ay, novo_subs)
if novo_subs is None:
return None
return novo_subs
return None
def unificar_variavel(var: Variavel, x: Any, subs: Dict[str, Any]) -> Optional[Dict[str, Any]]:
if var.nome in subs:
return unificar(subs[var.nome], x, subs)
novo_subs = dict(subs)
novo_subs[var.nome] = x
return novo_subs
# 3. Base de Conhecimento e Motor de Inferência Dedutivo
class BaseConhecimento:
def __init__(self):
self.fatos: List[Termo] = []
self.regras: List[Regra] = []
def declarar_fato(self, fato: Termo):
self.fatos.append(fato)
def declarar_regra(self, regra: Regra):
self.regras.append(regra)
def consultar(self, meta: Termo, subs: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
if subs is None:
subs = {}
resultados = []
# Tentar unificar diretamente com fatos
for fato in self.fatos:
s = unificar(meta, fato, subs)
if s is not None:
resultados.append(s)
# Tentar provar via regras com backtracking
for regra in self.regras:
s = unificar(meta, regra.cabecalho, subs)
if s is not None:
# Provar todas as premissas em cadeia
solucoes_correntes = [s]
for premissa in regra.premissas:
novas_solucoes = []
for sol in solucoes_correntes:
premissa_resolvida = self._aplicar_subs(premissa, sol)
sub_resultados = self.consultar(premissa_resolvida, sol)
novas_solucoes.extend(sub_resultados)
solucoes_correntes = novas_solucoes
resultados.extend(solucoes_correntes)
return resultados
def _aplicar_subs(self, termo: Termo, subs: Dict[str, Any]) -> Termo:
novos_args = [subs.get(arg.nome, arg) if isinstance(arg, Variavel) else arg for arg in termo.argumentos]
return Termo(termo.predicado, *novos_args)
# 4. Demonstração Prática: Regras Comerciais do Carrinho
if __name__ == "__main__":
bc = BaseConhecimento()
# Declaração de Fatos (Estado do E-commerce)
bc.declarar_fato(Termo("cliente_categoria", "carlos_silva", "vip"))
bc.declarar_fato(Termo("cliente_categoria", "ana_paula", "convencional"))
bc.declarar_fato(Termo("cliente_uf", "carlos_silva", "SP"))
bc.declarar_fato(Termo("cliente_uf", "ana_paula", "RJ"))
bc.declarar_fato(Termo("total_carrinho", "carlos_silva", 450.0))
bc.declarar_fato(Termo("total_carrinho", "ana_paula", 180.0))
# Declaração de Regras de Negócio
# Regra 1: Cliente VIP tem direito a frete grátis nacional
C = Variavel("C")
bc.declarar_regra(Regra(
cabecalho=Termo("frete_gratis", C),
premissas=[Termo("cliente_categoria", C, "vip")]
))
# Regra 2: Isenção fiscal regional para moradores de SP
bc.declarar_regra(Regra(
cabecalho=Termo("isencao_imposto_regional", C),
premissas=[Termo("cliente_uf", C, "SP")]
))
print("=" * 65)
print("🧠 DEDUÇÃO LÓGICA E INFERÊNCIA DO CARRINHO DE COMPRAS")
print("=" * 65)
# Consulta 1: Quem tem direito a Frete Grátis?
X = Variavel("ClienteElegivel")
solucoes_frete = bc.consultar(Termo("frete_gratis", X))
print(f"Consulta: ?- frete_gratis(?ClienteElegivel).")
for sol in solucoes_frete:
print(f" -> Resultado Provado: ?ClienteElegivel = {sol.get('ClienteElegivel')}")
# Consulta 2: Quem tem direito a Isenção Regional?
solucoes_imposto = bc.consultar(Termo("isencao_imposto_regional", X))
print(f"\nConsulta: ?- isencao_imposto_regional(?ClienteElegivel).")
for sol in solucoes_imposto:
print(f" -> Resultado Provado: ?ClienteElegivel = {sol.get('ClienteElegivel')}")
2. Barramento Reativo de Eventos com Contrapressão (Reactive Stream)
Neste exemplo, demonstramos o paradigma reativo (Reactive Streams). Os eventos de compra chegam de forma assíncrona no tempo. Para proteger o serviço de checkout de sobrecarga em eventos de pico (ex.: Black Friday), implementamos controle de contrapressão (backpressure) com buffer delimitado:
# barramento_reativo_checkout.py
import time
from typing import Callable, List, Any
from dataclasses import dataclass
@dataclass(frozen=True)
class EventoCarrinho:
id_pedido: str
cliente: str
valor_total: float
timestamp: float
class ReativoStreamCheckout:
"""Implementa um stream reativo push com operador de throttling/buffer para contrapressão."""
def __init__(self, limite_buffer: int = 3):
self._assinantes: List[Callable[[EventoCarrinho], None]] = []
self._buffer: List[EventoCarrinho] = []
self._limite_buffer = limite_buffer
self._descartes = 0
def assinar(self, observador: Callable[[EventoCarrinho], None]):
"""Registra um observador para reagir aos eventos emitidos."""
self._assinantes.append(observador)
def emitir(self, evento: EventoCarrinho):
"""Dispara um evento no fluxo aplicando estratégia de contenção Drop-Tail."""
if len(self._buffer) >= self._limite_buffer:
self._descartes += 1
print(f" ⚠️ [CONTRAPRESSÃO / DROP] Buffer saturado! Evento {evento.id_pedido} descartado para proteger o sistema.")
return
self._buffer.append(evento)
print(f" 📥 [BUFFER ENFILEIRADO] Evento {evento.id_pedido} aceito. (Ocupação: {len(self._buffer)}/{self._limite_buffer})")
def drenar_processamento(self):
"""Consome o buffer simulando a capacidade máxima do serviço de liquidação."""
print("\n⚡ [STREAM CONSUMER] Iniciando drenagem reativa do buffer...")
while self._buffer:
evento = self._buffer.pop(0)
for assinante in self._assinantes:
assinante(evento)
print("⚡ [STREAM CONSUMER] Buffer 100% estabilizado.")
# Observadores / Serviços Reativos Independentes
def servico_faturamento(ev: EventoCarrinho):
print(f" 💳 [Faturamento] Liquidando R$ {ev.valor_total:.2f} do pedido #{ev.id_pedido} ({ev.cliente})")
def servico_logistica(ev: EventoCarrinho):
print(f" 📦 [Logística] Separando produtos do pedido #{ev.id_pedido} no armazém.")
# Demonstração Prática
if __name__ == "__main__":
stream = ReativoStreamCheckout(limite_buffer=3)
# Conectando consumidores aos eventos
stream.assinar(servico_faturamento)
stream.assinar(servico_logistica)
print("=" * 65)
print("⚡ SIMULAÇÃO DE FLUXO REATIVO COM CONTRAPRESSÃO (BURST)")
print("=" * 65)
# Disparo em rajada (Burst) de 5 transações em milissegundos
eventos = [
EventoCarrinho("PED-001", "Marina Costa", 450.00, time.time()),
EventoCarrinho("PED-002", "Lucas Rocha", 120.00, time.time()),
EventoCarrinho("PED-003", "Beatriz Lima", 890.00, time.time()),
EventoCarrinho("PED-004", "Fernando Dias", 2100.00, time.time()), # Deverá sofrer Drop
EventoCarrinho("PED-005", "Juliana Prado", 340.00, time.time()), # Deverá sofrer Drop
]
print("\n[FASE 1: Rajada de Checkout Recebida no Servidor]")
for ev in eventos:
stream.emitir(ev)
print("\n[FASE 2: Consumidores Processam Fila com Resiliência]")
stream.drenar_processamento()
Saída Esperada dos Scripts:
=================================================================
🧠 DEDUÇÃO LÓGICA E INFERÊNCIA DO CARRINHO DE COMPRAS
=================================================================
Consulta: ?- frete_gratis(?ClienteElegivel).
-> Resultado Provado: ?ClienteElegivel = carlos_silva
Consulta: ?- isencao_imposto_regional(?ClienteElegivel).
-> Resultado Provado: ?ClienteElegivel = carlos_silva
=================================================================
⚡ SIMULAÇÃO DE FLUXO REATIVO COM CONTRAPRESSÃO (BURST)
=================================================================
[FASE 1: Rajada de Checkout Recebida no Servidor]
📥 [BUFFER ENFILEIRADO] Evento PED-001 aceito. (Ocupação: 1/3)
📥 [BUFFER ENFILEIRADO] Evento PED-002 aceito. (Ocupação: 2/3)
📥 [BUFFER ENFILEIRADO] Evento PED-003 aceito. (Ocupação: 3/3)
⚠️ [CONTRAPRESSÃO / DROP] Buffer saturado! Evento PED-004 descartado para proteger o sistema.
⚠️ [CONTRAPRESSÃO / DROP] Buffer saturado! Evento PED-005 descartado para proteger o sistema.
[FASE 2: Consumidores Processam Fila com Resiliência]
⚡ [STREAM CONSUMER] Iniciando drenagem reativa do buffer...
💳 [Faturamento] Liquidando R$ 450.00 do pedido #PED-001 (Marina Costa)
📦 [Logística] Separando produtos do pedido #PED-001 no armazém.
💳 [Faturamento] Liquidando R$ 120.00 do pedido #PED-002 (Lucas Rocha)
📦 [Logística] Separando produtos do pedido #PED-002 no armazém.
💳 [Faturamento] Liquidando R$ 890.00 do pedido #PED-003 (Beatriz Lima)
📦 [Logística] Separando produtos do pedido #PED-003 no armazém.
⚡ [STREAM CONSUMER] Buffer 100% estabilizado.