Padrões de Design em Python para Arquitetura Limpa
Construa aplicativos Python mantíveis com padrões de design SOLID
A Clean Architecture revolucionou a forma como os desenvolvedores constroem aplicações escaláveis e mantidas, enfatizando a separação de preocupações e o gerenciamento de dependências.
Em Python, esses princípios combinam-se com a natureza dinâmica da linguagem para criar sistemas flexíveis e testáveis que evoluem junto com os requisitos de negócio, sem se transformar em dívida técnica.
Compreendendo a Clean Architecture em Python
A Clean Architecture, introduzida por Robert C. Martin (Uncle Bob), organiza o software em camadas concêntricas onde as dependências apontam para o interior, em direção à lógica de negócio central. Este padrão arquitetural garante que as regras de negócio críticas da sua aplicação permaneçam independentes de frameworks, bancos de dados e serviços externos.
A Filosofia Central
O princípio fundamental é simples, porém poderoso: a lógica de negócio não deve depender da infraestrutura. Suas entidades de domínio, casos de uso e regras de negócio devem funcionar independentemente de você estar usando PostgreSQL ou MongoDB, FastAPI ou Flask, AWS ou Azure.
Em Python, essa filosofia alinha-se perfeitamente com o “duck typing” e a programação orientada a protocolos da linguagem, permitindo uma separação limpa sem a cerimônia necessária em linguagens tipadas estaticamente.
As Quatro Camadas da Clean Architecture
Camada de Entidades (Domínio): Objetos de negócio puros com regras de negócio de nível empresarial. Estes são POJOs (Plain Old Python Objects) sem dependências externas.
Camada de Casos de Uso (Aplicação): Regras de negócio específicas da aplicação que orquestram o fluxo de dados entre entidades e serviços externos.
Camada de Adaptadores de Interface: Converte dados entre o formato mais conveniente para casos de uso e entidades, e o formato exigido por agências externas.
Camada de Frameworks e Drivers: Todos os detalhes externos, como bancos de dados, frameworks web e APIs externas.
Princípios SOLID em Python
Os princípios SOLID formam a fundação da clean architecture. Vamos explorar como cada princípio se manifesta em Python. Para uma visão geral abrangente de padrões de design em Python, consulte o Guia de Padrões de Design em Python.
Princípio da Responsabilidade Única (SRP)
Cada classe deve ter apenas um motivo para mudar:
# Ruim: Múltiplas responsabilidades
class UserManager:
def create_user(self, user_data):
# Criar usuário
pass
def send_welcome_email(self, user):
# Enviar e-mail
pass
def log_creation(self, user):
# Registrar em arquivo
pass
# Bom: Responsabilidades separadas
class UserService:
def __init__(self, repository, email_service, logger):
self.repository = repository
self.email_service = email_service
self.logger = logger
def create_user(self, user_data):
user = User(**user_data)
self.repository.save(user)
self.email_service.send_welcome(user)
self.logger.info(f"Usuário criado: {user.id}")
return user
Princípio Aberto/Fechado (OCP)
Entidades de software devem ser abertas para extensão, mas fechadas para modificação:
from abc import ABC, abstractmethod
from typing import Protocol
# Usando Protocol (Python 3.8+)
class PaymentProcessor(Protocol):
def process_payment(self, amount: float) -> bool:
...
class CreditCardProcessor:
def process_payment(self, amount: float) -> bool:
# Lógica de cartão de crédito
return True
class PayPalProcessor:
def process_payment(self, amount: float) -> bool:
# Lógica do PayPal
return True
# Facilmente extensível sem modificar o código existente
class CryptoProcessor:
def process_payment(self, amount: float) -> bool:
# Lógica de criptomoeda
return True
Princípio da Substituição de Liskov (LSP)
Objetos devem ser substituíveis por seus subtipos sem quebrar o programa:
from abc import ABC, abstractmethod
class DataStore(ABC):
@abstractmethod
def save(self, key: str, value: str) -> None:
pass
@abstractmethod
def get(self, key: str) -> str:
pass
class PostgreSQLStore(DataStore):
def save(self, key: str, value: str) -> None:
# Implementação PostgreSQL
pass
def get(self, key: str) -> str:
# Implementação PostgreSQL
return ""
class RedisStore(DataStore):
def save(self, key: str, value: str) -> None:
# Implementação Redis
pass
def get(self, key: str) -> str:
# Implementação Redis
return ""
# Ambos podem ser usados de forma intercambiável
def process_data(store: DataStore, key: str, value: str):
store.save(key, value)
return store.get(key)
Princípio da Segregação de Interface (ISP)
Clientes não devem ser forçados a depender de interfaces que não utilizam:
# Ruim: Interface gorduda
class Worker(ABC):
@abstractmethod
def work(self): pass
@abstractmethod
def eat(self): pass
@abstractmethod
def sleep(self): pass
# Bom: Interfaces segregadas
class Workable(Protocol):
def work(self) -> None: ...
class Eatable(Protocol):
def eat(self) -> None: ...
class Human:
def work(self) -> None:
print("Trabalhando")
def eat(self) -> None:
print("Comendo")
class Robot:
def work(self) -> None:
print("Trabalhando")
# Método eat não é necessário
Princípio da Inversão de Dependência (DIP)
Módulos de alto nível não devem depender de módulos de baixo nível. Ambos devem depender de abstrações:
from typing import Protocol
# Abstração
class EmailSender(Protocol):
def send(self, to: str, subject: str, body: str) -> None:
...
# Módulo de baixo nível
class SMTPEmailSender:
def send(self, to: str, subject: str, body: str) -> None:
# Implementação SMTP
pass
# Módulo de alto nível depende da abstração
class UserRegistrationService:
def __init__(self, email_sender: EmailSender):
self.email_sender = email_sender
def register(self, email: str, name: str):
# Lógica de registro
self.email_sender.send(
to=email,
subject="Bem-vindo!",
body=f"Olá {name}"
)
Padrão Repository: Abstraindo o Acesso a Dados
O Padrão Repository fornece uma interface semelhante a coleção para acessar objetos de domínio, ocultando os detalhes de armazenamento de dados.
Implementação Básica do Repository
from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from uuid import UUID, uuid4
@dataclass
class User:
id: UUID
email: str
name: str
is_active: bool = True
class UserRepository(ABC):
@abstractmethod
def save(self, user: User) -> User:
pass
@abstractmethod
def get_by_id(self, user_id: UUID) -> Optional[User]:
pass
@abstractmethod
def get_by_email(self, email: str) -> Optional[User]:
pass
@abstractmethod
def list_all(self) -> List[User]:
pass
@abstractmethod
def delete(self, user_id: UUID) -> bool:
pass
Implementação com SQLAlchemy
from sqlalchemy import create_engine, Column, String, Boolean
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
Base = declarative_base()
class UserModel(Base):
__tablename__ = 'users'
id = Column(PGUUID(as_uuid=True), primary_key=True)
email = Column(String, unique=True, nullable=False)
name = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
class SQLAlchemyUserRepository(UserRepository):
def __init__(self, session: Session):
self.session = session
def save(self, user: User) -> User:
user_model = UserModel(
id=user.id,
email=user.email,
name=user.name,
is_active=user.is_active
)
self.session.add(user_model)
self.session.commit()
return user
def get_by_id(self, user_id: UUID) -> Optional[User]:
user_model = self.session.query(UserModel).filter(
UserModel.id == user_id
).first()
if not user_model:
return None
return User(
id=user_model.id,
email=user_model.email,
name=user_model.name,
is_active=user_model.is_active
)
def get_by_email(self, email: str) -> Optional[User]:
user_model = self.session.query(UserModel).filter(
UserModel.email == email
).first()
if not user_model:
return None
return User(
id=user_model.id,
email=user_model.email,
name=user_model.name,
is_active=user_model.is_active
)
def list_all(self) -> List[User]:
users = self.session.query(UserModel).all()
return [
User(
id=u.id,
email=u.email,
name=u.name,
is_active=u.is_active
)
for u in users
]
def delete(self, user_id: UUID) -> bool:
result = self.session.query(UserModel).filter(
UserModel.id == user_id
).delete()
self.session.commit()
return result > 0
Repository In-Memory para Testes
class InMemoryUserRepository(UserRepository):
def __init__(self):
self.users: dict[UUID, User] = {}
def save(self, user: User) -> User:
self.users[user.id] = user
return user
def get_by_id(self, user_id: UUID) -> Optional[User]:
return self.users.get(user_id)
def get_by_email(self, email: str) -> Optional[User]:
for user in self.users.values():
if user.email == email:
return user
return None
def list_all(self) -> List[User]:
return list(self.users.values())
def delete(self, user_id: UUID) -> bool:
if user_id in self.users:
del self.users[user_id]
return True
return False
Camada de Serviço: Orquestrando a Lógica de Negócio
A Camada de Serviço implementa casos de uso e orquestra o fluxo entre repositórios, serviços externos e lógica de domínio.
from typing import Optional
from uuid import uuid4
class UserAlreadyExistsError(Exception):
pass
class UserNotFoundError(Exception):
pass
class UserService:
def __init__(
self,
user_repository: UserRepository,
email_service: EmailSender,
event_publisher: 'EventPublisher'
):
self.user_repository = user_repository
self.email_service = email_service
self.event_publisher = event_publisher
def register_user(self, email: str, name: str) -> User:
# Verifica se o usuário existe
existing_user = self.user_repository.get_by_email(email)
if existing_user:
raise UserAlreadyExistsError(f"Usuário com e-mail {email} já existe")
# Cria novo usuário
user = User(
id=uuid4(),
email=email,
name=name,
is_active=True
)
# Salva no repositório
user = self.user_repository.save(user)
# Envia e-mail de boas-vindas
self.email_service.send(
to=user.email,
subject="Bem-vindo!",
body=f"Olá {user.name}, bem-vindo à nossa plataforma!"
)
# Publica evento
self.event_publisher.publish('user.registered', {
'user_id': str(user.id),
'email': user.email
})
return user
def deactivate_user(self, user_id: UUID) -> User:
user = self.user_repository.get_by_id(user_id)
if not user:
raise UserNotFoundError(f"Usuário {user_id} não encontrado")
user.is_active = False
user = self.user_repository.save(user)
self.event_publisher.publish('user.deactivated', {
'user_id': str(user.id)
})
return user
Injeção de Dependência em Python
A natureza dinâmica do Python torna a injeção de dependência direta, sem exigir frameworks pesados.
Injeção via Construtor
class OrderService:
def __init__(
self,
order_repository: 'OrderRepository',
payment_processor: PaymentProcessor,
notification_service: 'NotificationService'
):
self.order_repository = order_repository
self.payment_processor = payment_processor
self.notification_service = notification_service
def place_order(self, order_data: dict):
# Usa dependências injetadas
pass
Container de Dependência Simples
from typing import Dict, Type, Callable, Any
class Container:
def __init__(self):
self._services: Dict[Type, Callable] = {}
self._singletons: Dict[Type, Any] = {}
def register(self, interface: Type, factory: Callable):
self._services[interface] = factory
def register_singleton(self, interface: Type, instance: Any):
self._singletons[interface] = instance
def resolve(self, interface: Type):
if interface in self._singletons:
return self._singletons[interface]
factory = self._services.get(interface)
if factory:
return factory(self)
raise ValueError(f"Nenhum registro encontrado para {interface}")
# Uso
def create_container() -> Container:
container = Container()
# Registra serviços
container.register_singleton(
Session,
sessionmaker(bind=create_engine('postgresql://...'))()
)
container.register(
UserRepository,
lambda c: SQLAlchemyUserRepository(c.resolve(Session))
)
container.register(
EmailSender,
lambda c: SMTPEmailSender()
)
container.register(
UserService,
lambda c: UserService(
c.resolve(UserRepository),
c.resolve(EmailSender),
c.resolve(EventPublisher)
)
)
return container
Arquitetura Hexagonal (Ports and Adapters)
A Arquitetura Hexagonal coloca a lógica de negócio no centro, com adaptadores lidando com a comunicação externa.
Definindo Ports (Interfaces)
# Input Port (Primário)
class CreateUserUseCase(Protocol):
def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
...
# Output Port (Secundário)
class UserPersistencePort(Protocol):
def save(self, user: User) -> User:
...
def find_by_email(self, email: str) -> Optional[User]:
...
Implementando Adapters
from pydantic import BaseModel, EmailStr
# Input Adapter (REST API)
from fastapi import FastAPI, Depends, HTTPException
class CreateUserRequest(BaseModel):
email: EmailStr
name: str
class CreateUserResponse(BaseModel):
id: str
email: str
name: str
app = FastAPI()
@app.post("/users", response_model=CreateUserResponse)
def create_user(
request: CreateUserRequest,
user_service: UserService = Depends(get_user_service)
):
try:
user = user_service.register_user(
email=request.email,
name=request.name
)
return CreateUserResponse(
id=str(user.id),
email=user.email,
name=user.name
)
except UserAlreadyExistsError as e:
raise HTTPException(status_code=400, detail=str(e))
# Output Adapter (Banco de Dados)
# Já implementado como SQLAlchemyUserRepository
Padrões de Design Orientado a Domínio
Value Objects (Objetos de Valor)
Objetos imutáveis definidos por seus atributos:
from dataclasses import dataclass
from typing import Pattern
import re
@dataclass(frozen=True)
class Email:
value: str
EMAIL_PATTERN: Pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
def __post_init__(self):
if not self.EMAIL_PATTERN.match(self.value):
raise ValueError(f"E-mail inválido: {self.value}")
def __str__(self):
return self.value
@dataclass(frozen=True)
class Money:
amount: float
currency: str
def __post_init__(self):
if self.amount < 0:
raise ValueError("O valor não pode ser negativo")
if self.currency not in ['USD', 'EUR', 'GBP']:
raise ValueError(f"Moeda não suportada: {self.currency}")
def add(self, other: 'Money') -> 'Money':
if self.currency != other.currency:
raise ValueError("Não é possível adicionar moedas diferentes")
return Money(self.amount + other.amount, self.currency)
Aggregates (Agregados)
Agrupamento de objetos de domínio tratados como uma única unidade:
from dataclasses import dataclass, field
from typing import List
from datetime import datetime
@dataclass
class OrderItem:
product_id: UUID
quantity: int
price: Money
def total(self) -> Money:
return Money(
self.price.amount * self.quantity,
self.price.currency
)
@dataclass
class Order:
id: UUID
customer_id: UUID
items: List[OrderItem] = field(default_factory=list)
status: str = "pending"
created_at: datetime = field(default_factory=datetime.now)
def add_item(self, product_id: UUID, quantity: int, price: Money):
item = OrderItem(product_id, quantity, price)
self.items.append(item)
def remove_item(self, product_id: UUID):
self.items = [
item for item in self.items
if item.product_id != product_id
]
def total(self) -> Money:
if not self.items:
return Money(0, "USD")
return sum(
(item.total() for item in self.items),
Money(0, self.items[0].price.currency)
)
def confirm(self):
if not self.items:
raise ValueError("Não é possível confirmar pedido vazio")
if self.status != "pending":
raise ValueError("Pedido já processado")
self.status = "confirmed"
Domain Events (Eventos de Domínio)
Eventos de domínio permitem acoplamento fraco entre componentes e suportam arquiteturas orientadas a eventos. Para sistemas orientados a eventos em escala de produção, considere implementar streaming de eventos com serviços como AWS Kinesis — consulte Building Event-Driven Microservices with AWS Kinesis para um guia detalhado.
from dataclasses import dataclass
from datetime import datetime
from typing import List, Callable
@dataclass
class DomainEvent:
occurred_at: datetime = field(default_factory=datetime.now)
@dataclass
class OrderConfirmed(DomainEvent):
order_id: UUID
customer_id: UUID
total: Money
class EventPublisher:
def __init__(self):
self._handlers: Dict[Type, List[Callable]] = {}
def subscribe(self, event_type: Type, handler: Callable):
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
def publish(self, event: DomainEvent):
event_type = type(event)
handlers = self._handlers.get(event_type, [])
for handler in handlers:
handler(event)
Recursos Modernos do Python para Clean Architecture
Os recursos modernos do Python tornam a implementação de clean architecture mais elegante e segura em termos de tipos. Se você precisa de uma referência rápida para a sintaxe e recursos do Python, confira a Python Cheatsheet.
Type Hints e Protocols
from typing import Protocol, runtime_checkable
@runtime_checkable
class Serializable(Protocol):
def to_dict(self) -> dict:
...
@classmethod
def from_dict(cls, data: dict) -> 'Serializable':
...
def serialize(obj: Serializable) -> dict:
return obj.to_dict()
Pydantic para Validação
from pydantic import BaseModel, Field, validator
from typing import Optional
class CreateUserDTO(BaseModel):
email: EmailStr
name: str = Field(..., min_length=2, max_length=100)
age: Optional[int] = Field(None, ge=0, le=150)
@validator('name')
def name_must_not_contain_numbers(cls, v):
if any(char.isdigit() for char in v):
raise ValueError('Nome não pode conter números')
return v
class Config:
frozen = True # Tornar imutável
Async/Await para Operações de I/O
A sintaxe async/await do Python é particularmente poderosa para operações orientadas a I/O na clean architecture, permitindo interações não bloqueantes com bancos de dados e serviços externos. Ao implantar aplicações Python em plataformas serverless, entender as características de desempenho torna-se crucial — veja AWS lambda performance: JavaScript vs Python vs Golang para insights sobre otimização de funções serverless em Python.
from typing import List
import asyncio
class AsyncUserRepository(ABC):
@abstractmethod
async def save(self, user: User) -> User:
pass
@abstractmethod
async def get_by_id(self, user_id: UUID) -> Optional[User]:
pass
class AsyncUserService:
def __init__(self, repository: AsyncUserRepository):
self.repository = repository
async def register_user(self, email: str, name: str) -> User:
user = User(id=uuid4(), email=email, name=name)
return await self.repository.save(user)
async def get_users_batch(self, user_ids: List[UUID]) -> List[User]:
tasks = [self.repository.get_by_id(uid) for uid in user_ids]
results = await asyncio.gather(*tasks)
return [u for u in results if u is not None]
Melhores Práticas de Estrutura de Projeto
A organização adequada do projeto é essencial para manter a clean architecture. Antes de configurar a estrutura do seu projeto, certifique-se de usar ambientes virtuais Python para isolamento de dependências. A venv Cheatsheet cobre tudo o que você precisa saber sobre gerenciamento de ambientes virtuais. Para projetos Python modernos, considere usar uv - New Python Package, Project, and Environment Manager, que fornece gerenciamento de pacotes e configuração de projetos mais rápidos.
my_application/
├── domain/ # Regras de negócio da empresa
│ ├── __init__.py
│ ├── entities/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── order.py
│ ├── value_objects/
│ │ ├── __init__.py
│ │ ├── email.py
│ │ └── money.py
│ ├── events/
│ │ ├── __init__.py
│ │ └── user_events.py
│ └── exceptions.py
│
├── application/ # Regras de negócio da aplicação
│ ├── __init__.py
│ ├── use_cases/
│ │ ├── __init__.py
│ │ ├── create_user.py
│ │ └── place_order.py
│ ├── services/
│ │ ├── __init__.py
│ │ └── user_service.py
│ └── ports/
│ ├── __init__.py
│ ├── repositories.py
│ └── external_services.py
│
├── infrastructure/ # Interfaces externas
│ ├── __init__.py
│ ├── persistence/
│ │ ├── __init__.py
│ │ ├── sqlalchemy/
│ │ │ ├── models.py
│ │ │ └── repositories.py
│ │ └── mongodb/
│ │ └── repositories.py
│ ├── messaging/
│ │ ├── __init__.py
│ │ └── rabbitmq_publisher.py
│ ├── external_services/
│ │ ├── __init__.py
│ │ └── email_service.py
│ └── config.py
│
├── presentation/ # Camada UI/API
│ ├── __init__.py
│ ├── api/
│ │ ├── __init__.py
│ │ ├── dependencies.py
│ │ ├── routes/
│ │ │ ├── __init__.py
│ │ │ ├── users.py
│ │ │ └── orders.py
│ │ └── schemas/
│ │ ├── __init__.py
│ │ └── user_schemas.py
│ └── cli/
│ └── commands.py
│
├── tests/
│ ├── unit/
│ ├── integration/
│ └── e2e/
│
├── main.py # Ponto de entrada da aplicação
├── container.py # Configuração de injeção de dependência
├── pyproject.toml
└── README.md
Testando Clean Architecture
Testes Unitários de Lógica de Domínio
import pytest
from uuid import uuid4
def test_user_creation():
user = User(
id=uuid4(),
email="test@example.com",
name="Test User"
)
assert user.email == "test@example.com"
assert user.is_active is True
def test_order_total_calculation():
order = Order(id=uuid4(), customer_id=uuid4())
order.add_item(
uuid4(),
quantity=2,
price=Money(10.0, "USD")
)
order.add_item(
uuid4(),
quantity=1,
price=Money(5.0, "USD")
)
assert order.total().amount == 25.0
Testes de Integração com Repository
@pytest.fixture
def in_memory_repository():
return InMemoryUserRepository()
def test_user_repository_save_and_retrieve(in_memory_repository):
user = User(
id=uuid4(),
email="test@example.com",
name="Test User"
)
saved_user = in_memory_repository.save(user)
retrieved_user = in_memory_repository.get_by_id(user.id)
assert retrieved_user is not None
assert retrieved_user.email == user.email
Testando a Camada de Serviço
from unittest.mock import Mock
def test_user_registration():
# Arrange
mock_repository = Mock(spec=UserRepository)
mock_repository.get_by_email.return_value = None
mock_repository.save.return_value = User(
id=uuid4(),
email="test@example.com",
name="Test"
)
mock_email = Mock(spec=EmailSender)
mock_events = Mock(spec=EventPublisher)
service = UserService(mock_repository, mock_email, mock_events)
# Act
user = service.register_user("test@example.com", "Test")
# Assert
assert user.email == "test@example.com"
mock_repository.save.assert_called_once()
mock_email.send.assert_called_once()
mock_events.publish.assert_called_once()
Armadilhas Comuns e Como Evitá-las
Superengenharia
Não implemente clean architecture para aplicações CRUD simples. Comece simples e refatore à medida que a complexidade cresce.
Abstrações Vazadas
Garanta que as entidades de domínio não contenham anotações de banco de dados ou código específico de framework:
# Ruim
from sqlalchemy import Column
@dataclass
class User:
id: Column(Integer, primary_key=True) # Framework vazando para o domínio
# Bom
@dataclass
class User:
id: UUID # Objeto de domínio puro
Dependências Circulares
Use injeção de dependência e interfaces para quebrar dependências circulares entre camadas.
Ignorando o Contexto
Clean Architecture não é uma solução única para todos. Ajuste a rigorosidade das camadas com base no tamanho do projeto e na expertise da equipe.
Links Úteis
- Clean Architecture de Robert C. Martin
- Documentação de Type Hints do Python
- Documentação do Pydantic
- Documentação Oficial do FastAPI
- Documentação do SQLAlchemy ORM
- Biblioteca Dependency Injector
- Referência de Design Orientado a Domínio
- Padrões de Arquitetura com Python
- Blog de Martin Fowler sobre Arquitetura
- Guia de Padrões de Design em Python
- Python Cheatsheet
- venv Cheatsheet
- uv - New Python Package, Project, and Environment Manager
- AWS lambda performance: JavaScript vs Python vs Golang
- Building Event-Driven Microservices with AWS Kinesis
