Tutorial Prático: Arquitetura Hexagonal em Python (Ports & Adapters)
A Arquitetura Hexagonal, também conhecida como padrão Ports & Adapters (Portas e Adaptadores), foi proposta por Alistair Cockburn em 2005. O objetivo central é isolar as regras de negócio centrais (Domínio e Casos de Uso) de todas as tecnologias e detalhes externos (bancos de dados, frameworks web, filas, APIs terceiras, interfaces CLI).
Dessa forma, sua aplicação pode ser executada igualmente por uma API REST (FastAPI), uma interface de linha de comando (CLI), uma suíte de testes automatizados ou um script agendado, sem alterar uma única linha da regra de negócio.
1. Conceitos Fundamentais
A arquitetura organiza o software em três grandes áreas:
- Núcleo da Aplicação (Inside / Hexágono Interno):
- Domínio (Domain): Entidades puras, objetos de valor e regras de negócio invariáveis.
- Casos de Uso (Application / Use Cases): Fluxos de orquestração do sistema.
- Portas (Ports): Contratos abstratos (interfaces) que definem como o mundo exterior entra ou sai da aplicação.
- Adaptadores (Outside / Hexágono Externo):
- O código que converte chamadas externas para o formato do hexágono, ou vice-versa.
- Ponto de Composição (Composition Root):
- Onde os adaptadores são instanciados e conectados (injetados) nas portas dos casos de uso.
[ Usuário / Web API / CLI ]
│
▼ (Driver / Entrada)
┌───────────────────┐
│ Inbound Adapter │ (ex: FastAPI Route, CLI parser)
└─────────┬─────────┘
│
▼
╭───────────┴───────────╮
│ INBOUND PORT │ (ex: Interface do Caso de Uso)
│ ─────────────────── │
│ APPLICATION / │
│ CORE DOMAIN │ (Regras de negócio puras)
│ ─────────────────── │
│ OUTBOUND PORT │ (ex: Protocols de Repositório, Notificação)
╰───────────┬───────────╯
│
▼
┌─────────┴─────────┐
│ Outbound Adapter │ (ex: PostgreSQL, Firestore, SendGrid)
└─────────┬─────────┘
│
▼ (Driven / Saída)
[ Banco / Fila / Rede ]
Portas Primárias vs. Secundárias
| Tipo | Também chamado de | Papel | Exemplos em Python |
|---|---|---|---|
| Primárias (Inbound / Driving) | Portas de Entrada | Definem como a aplicação é acionada por atores externos. | Interfaces abstratas de Casos de Uso invocadas por controllers REST, CLI ou consumers de fila. |
| Secundárias (Outbound / Driven) | Portas de Saída | Definem como o núcleo busca ou envia dados para o exterior. | typing.Protocol ou abc.ABC para repositórios SQL, clientes de e-mail, buckets S3, logs corporativos. |
2. Estrutura de Pastas Recomendada
No Python moderno, recomenda-se a seguinte organização modular:
meu_projeto/
├── src/
│ └── fatura_app/
│ ├── domain/ # 1. NÚCLEO PURO (Zero dependências externas)
│ │ ├── exceptions.py
│ │ └── models.py # Entidades e Value Objects (dataclasses)
│ │
│ ├── ports/ # 2. CONTRATOS E INTERFACES
│ │ ├── inbound/ # Portas de Entrada (Use Cases)
│ │ │ └── processar_fatura.py
│ │ └── outbound/ # Portas de Saída (Repositórios, Gateways)
│ │ ├── fatura_repository.py
│ │ └── notification_service.py
│ │
│ ├── application/ # 3. CASOS DE USO (Implementação das portas de entrada)
│ │ └── processar_fatura_use_case.py
│ │
│ └── adapters/ # 4. ADAPTADORES EXTERNOS
│ ├── inbound/ # Como entram os dados
│ │ ├── cli.py # Adaptador de CLI (Argparse / Click / Typer)
│ │ └── api/ # Adaptador Web (FastAPI / Flask)
│ │ └── routes.py
│ └── outbound/ # Como saem os dados
│ ├── database/
│ │ ├── memory_fatura_repo.py # Para testes unitários
│ │ └── postgres_fatura_repo.py # Produção com SQLAlchemy
│ └── notifications/
│ ├── console_notifier.py # Local / Dev
│ └── sendgrid_notifier.py # Produção
│
├── tests/
│ ├── unit/ # Testam o núcleo usando adaptadores em memória
│ └── integration/ # Testam adaptadores reais (Postgres, etc.)
├── main.py # Composition Root (Instancia e conecta tudo)
└── pyproject.toml
3. Implementação Prática Passo a Passo
Vamos implementar um sistema de Processamento de Faturas.
Passo 1: Domínio (domain/models.py)
Entidades puras em Python usando dataclasses. Não dependem de ORMs, SQL ou bibliotecas externas.
# src/fatura_app/domain/models.py
from dataclasses import dataclass
from datetime import date
from enum import StrEnum
from uuid import UUID, uuid4
class StatusFatura(StrEnum):
PENDENTE = "pendente"
PAGA = "paga"
CANCELADA = "cancelada"
@dataclass
class Fatura:
cliente_id: str
valor: float
vencimento: date
id: UUID = uuid4()
status: StatusFatura = StatusFatura.PENDENTE
def marcar_como_paga(self) -> None:
if self.status == StatusFatura.CANCELADA:
raise ValueError("Não é possível pagar uma fatura cancelada.")
self.status = StatusFatura.PAGA
Passo 2: Portas de Saída / Outbound Ports (ports/outbound/)
Definidas usando typing.Protocol (tipagem estrutural estática). O núcleo decide como quer que os adaptadores se comportem.
# src/fatura_app/ports/outbound/fatura_repository.py
from typing import Protocol, Optional
from uuid import UUID
from fatura_app.domain.models import Fatura
class FaturaRepositoryPort(Protocol):
"""Porta de saída para persistência de faturas."""
def salvar(self, fatura: Fatura) -> None: ...
def buscar_por_id(self, fatura_id: UUID) -> Optional[Fatura]: ...
# src/fatura_app/ports/outbound/notification_service.py
from typing import Protocol
class NotificationPort(Protocol):
"""Porta de saída para disparos de mensagens e avisos."""
def notificar_pagamento(self, cliente_id: str, valor: float) -> None: ...
Passo 3: Porta de Entrada e Caso de Uso (application/)
A porta de entrada define o contrato do caso de uso, e a classe orquestra a lógica de negócio consumindo apenas as portas.
# src/fatura_app/ports/inbound/processar_fatura.py
from typing import Protocol
from uuid import UUID
class ProcessarPagamentoPort(Protocol):
"""Porta de entrada para efetivação de pagamentos de faturas."""
def executar(self, fatura_id: UUID) -> None: ...
# src/fatura_app/application/processar_fatura_use_case.py
from uuid import UUID
from fatura_app.ports.inbound.processar_fatura import ProcessarPagamentoPort
from fatura_app.ports.outbound.fatura_repository import FaturaRepositoryPort
from fatura_app.ports.outbound.notification_service import NotificationPort
class ProcessarPagamentoUseCase(ProcessarPagamentoPort):
"""
Núcleo da aplicação: depende APENAS de abstrações (ports).
Desconhece se o banco é Postgres, SQLite ou se a notificação é SMS ou E-mail.
"""
def __init__(
self,
repository: FaturaRepositoryPort,
notifier: NotificationPort
) -> None:
self._repo = repository
self._notifier = notifier
def executar(self, fatura_id: UUID) -> None:
fatura = self._repo.buscar_por_id(fatura_id)
if not fatura:
raise FileNotFoundError(f"Fatura {fatura_id} não encontrada.")
# Executa regra de negócio do domínio
fatura.marcar_como_paga()
# Persiste o novo estado via porta de saída
self._repo.salvar(fatura)
# Dispara evento/notificação via porta de saída
self._notifier.notificar_pagamento(fatura.cliente_id, fatura.valor)
Passo 4: Adaptadores de Saída / Outbound Adapters (adapters/outbound/)
Aqui conectamos tecnologias reais de infraestrutura às portas de saída.
Adaptador em Memória (ideal para testes rápidos):
# src/fatura_app/adapters/outbound/database/memory_fatura_repo.py
from typing import Dict, Optional
from uuid import UUID
from fatura_app.domain.models import Fatura
from fatura_app.ports.outbound.fatura_repository import FaturaRepositoryPort
class InMemoryFaturaRepository(FaturaRepositoryPort):
def __init__(self) -> None:
self._storage: Dict[UUID, Fatura] = {}
def salvar(self, fatura: Fatura) -> None:
self._storage[fatura.id] = fatura
def buscar_por_id(self, fatura_id: UUID) -> Optional[Fatura]:
return self._storage.get(fatura_id)
Adaptador de Notificação via Console (Dev/Sandbox):
# src/fatura_app/adapters/outbound/notifications/console_notifier.py
from fatura_app.ports.outbound.notification_service import NotificationPort
class ConsoleNotificationAdapter(NotificationPort):
def notificar_pagamento(self, cliente_id: str, valor: float) -> None:
print(f"[Notificação Console] Cliente '{cliente_id}' teve fatura de R$ {valor:.2f} liquidada.")
Passo 5: Adaptadores de Entrada / Inbound Adapters (adapters/inbound/)
Adaptadores que convertem entradas externas (ex: CLI, HTTP REST) em chamadas para o caso de uso.
Adaptador CLI:
# src/fatura_app/adapters/inbound/cli.py
from uuid import UUID
from fatura_app.ports.inbound.processar_fatura import ProcessarPagamentoPort
class FaturaCLIAdapter:
def __init__(self, use_case: ProcessarPagamentoPort) -> None:
self._use_case = use_case
def pagar_fatura_comando(self, raw_id: str) -> None:
try:
fatura_id = UUID(raw_id)
self._use_case.executar(fatura_id)
print(f"[CLI Sucesso] Fatura {raw_id} processada com sucesso!")
except Exception as exc:
print(f"[CLI Erro] Falha ao processar: {exc}")
Adaptador Web (FastAPI):
# src/fatura_app/adapters/inbound/api/routes.py
from uuid import UUID
from fastapi import APIRouter, HTTPException, Depends
from fatura_app.ports.inbound.processar_fatura import ProcessarPagamentoPort
router = APIRouter()
# Injeção da porta de entrada pelo container/framework
def get_use_case() -> ProcessarPagamentoPort:
# Resolvido pelo container de DI do FastAPI
raise NotImplementedError
@router.post("/faturas/{fatura_id}/pagar")
def pagar_fatura_endpoint(
fatura_id: UUID,
use_case: ProcessarPagamentoPort = Depends(get_use_case)
):
try:
use_case.executar(fatura_id)
return {"status": "sucesso", "fatura_id": str(fatura_id)}
except FileNotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc))
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
Passo 6: O Ponto de Composição (main.py)
No ponto de entrada da aplicação, nós instanciamos os adaptadores concretos e os "pluguamos" nos casos de uso:
# main.py
from datetime import date
from fatura_app.domain.models import Fatura
from fatura_app.application.processar_fatura_use_case import ProcessarPagamentoUseCase
from fatura_app.adapters.outbound.database.memory_fatura_repo import InMemoryFaturaRepository
from fatura_app.adapters.outbound.notifications.console_notifier import ConsoleNotificationAdapter
from fatura_app.adapters.inbound.cli import FaturaCLIAdapter
def main():
# 1. Configurar adaptadores de infraestrutura (Outbound Adapters)
repo = InMemoryFaturaRepository()
notifier = ConsoleNotificationAdapter()
# 2. Criar massa de dados inicial para demonstração
fatura_exemplo = Fatura(cliente_id="cliente-corp-101", valor=1450.75, vencimento=date.today())
repo.salvar(fatura_exemplo)
print(f"Fatura criada: {fatura_exemplo.id} (Status: {fatura_exemplo.status.value})")
# 3. Montar caso de uso conectando as portas de saída
caso_de_uso = ProcessarPagamentoUseCase(repository=repo, notifier=notifier)
# 4. Inicializar adaptador de entrada (Inbound Adapter)
cli = FaturaCLIAdapter(use_case=caso_de_uso)
# 5. Executar o fluxo
print("\n--- Executando via Adaptador CLI ---")
cli.pagar_fatura_comando(str(fatura_exemplo.id))
# Verificar estado no repositório
fatura_atualizada = repo.buscar_por_id(fatura_exemplo.id)
print(f"Status atualizado da fatura no banco: {fatura_atualizada.status.value}")
if __name__ == "__main__":
main()
4. Como Testar o Núcleo Sem Banco de Dados
A maior vantagem da arquitetura hexagonal é a facilidade de escrever testes unitários velozes sem depender de banco de dados, Docker ou serviços de nuvem:
# tests/unit/test_processar_fatura.py
from datetime import date
from uuid import uuid4
import pytest
from fatura_app.domain.models import Fatura, StatusFatura
from fatura_app.application.processar_fatura_use_case import ProcessarPagamentoUseCase
class StubRepository:
def __init__(self):
self.faturas = {}
self.salvou = False
def buscar_por_id(self, fatura_id):
return self.faturas.get(fatura_id)
def salvar(self, fatura):
self.faturas[fatura.id] = fatura
self.salvou = True
class SpyNotifier:
def __init__(self):
self.mensagens_enviadas = []
def notificar_pagamento(self, cliente_id, valor):
self.mensagens_enviadas.append((cliente_id, valor))
def test_deve_processar_pagamento_e_notificar():
repo = StubRepository()
notifier = SpyNotifier()
fatura = Fatura(cliente_id="cli-1", valor=500.0, vencimento=date.today())
repo.salvar(fatura)
use_case = ProcessarPagamentoUseCase(repository=repo, notifier=notifier)
use_case.executar(fatura.id)
# Asserções
assert repo.buscar_por_id(fatura.id).status == StatusFatura.PAGA
assert len(notifier.mensagens_enviadas) == 1
assert notifier.mensagens_enviadas[0] == ("cli-1", 500.0)
def test_deve_lancar_erro_quando_fatura_nao_existir():
repo = StubRepository()
notifier = SpyNotifier()
use_case = ProcessarPagamentoUseCase(repository=repo, notifier=notifier)
with pytest.raises(FileNotFoundError):
use_case.executar(uuid4())
5. Comparativo: Arquitetura Hexagonal vs. Arquitetura em Camadas Tradicional
| Aspecto | Camadas Tradicional (MVC / 3-Tier) | Hexagonal (Ports & Adapters) |
|---|---|---|
| Direção da Dependência | Apresentação Negócio Banco de Dados | Adaptadores Externos Portas Domínio Central |
| Dependência do Banco | A lógica de negócios depende diretamente de tabelas/ORMs | O banco depende da porta definida pelo domínio (Inversão de Dependência) |
| Facilidade de Testes | Alta dependência de bancos reais ou mocks complexos | Testes unitários do núcleo rodam instantaneamente com stubs simples |
| Troca de Tecnologia | Difícil: migrar de Postgres para NoSQL exige mexer no Service | Fácil: cria-se apenas um novo adaptador que satisfaça a porta |
| Curva de Aprendizado | Baixa a moderada | Moderada (requer disciplina na separação de portas e adaptadores) |
Resumo e Boas Práticas
- Domínio Imaculado: O diretório
domain/nunca deve importar bibliotecas de terceiros como SQLAlchemy, Django, FastAPI, Boto3 ou Pydantic. Use tipos e dataclasses nativos do Python. - Definição de Portas: Em Python moderno, prefira
typing.Protocolem vez de herança múltipla comabc.ABC, pois o duck typing estrutural facilita a criação de mocks e desacopla as classes concretas. - Casos de Uso Coesos: Cada caso de uso deve ter um único propósito público (ex.: método
executar()ou__call__()). - Composition Root Único: Mantenha a criação dos objetos e a resolução de dependências concentrada em pontos específicos (ex.:
main.py, factories ou providers de containers de DI).