Padrões de Design Python para Arquitetura Limpa

# 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

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

# Fácil de estender sem modificar o código existente
class CryptoProcessor:
    def process_payment(self, amount: float) -> bool:
        # Lógica de criptomoeda
        return True

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 do PostgreSQL
        pass
    
    def get(self, key: str) -> str:
        # Implementação do PostgreSQL
        return ""

class RedisStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # Implementação do Redis
        pass
    
    def get(self, key: str) -> str:
        # Implementação do 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)

# Ruim: Interface pesada
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 de comer não necessário

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}"
        )

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

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

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

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:
        # Verificar se o usuário já existe
        existing_user = self.user_repository.get_by_email(email)
        if existing_user:
            raise UserAlreadyExistsError(f"Usuário com e-mail {email} já existe")
        
        # Criar novo usuário
        user = User(
            id=uuid4(),
            email=email,
            name=name,
            is_active=True
        )
        
        # Salvar no repositório
        user = self.user_repository.save(user)
        
        # Enviar e-mail de boas-vindas
        self.email_service.send(
            to=user.email,
            subject="Bem-vindo!",
            body=f"Olá {user.name}, bem-vindo à nossa plataforma!"
        )
        
        # Publicar 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

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):
        # Usar dependências injetadas
        pass

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"Nenhuma inscrição encontrada para {interface}")

# Uso
def create_container() -> Container:
    container = Container()
    
    # Registrar 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

# Porta de Entrada (Primária)
class CreateUserUseCase(Protocol):
    def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
        ...

# Porta de Saída (Secundária)
class UserPersistencePort(Protocol):
    def save(self, user: User) -> User:
        ...
    
    def find_by_email(self, email: str) -> Optional[User]:
        ...

from pydantic import BaseModel, EmailStr

# Adaptador de Entrada (API REST)
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))

# Adaptador de Saída (Banco de Dados)
# Já implementado como SQLAlchemyUserRepository

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("Quantia não pode ser negativa")
        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)

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 um pedido vazio")
        if self.status != "pending":
            raise ValueError("Pedido já processado")
        self.status = "confirmed"

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)

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

```python
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('Name cannot contain numbers')
        return v
    
    class Config:
        frozen = True  # Torna imutável
`

### Async/Await para Operações de E/S

A sintaxe async/await do Python é particularmente poderosa para operações limitadas por E/S na arquitetura limpa, permitindo interações não bloqueantes com bancos de dados e serviços externos. Ao implantar aplicações Python em plataformas sem servidor, entender as características de desempenho se torna crucial—veja [AWS lambda performance: JavaScript vs Python vs Golang](https://www.glukhov.org/pt/post/2024/08/aws-lambda-golang-python-js/ "Desenvolvendo aplicações sem servidor na AWS com sam usando JavaScript, Python e Golang") para insights sobre otimização de funções Python sem servidor.

```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]
`

## Boas Práticas para Estrutura de Projeto

Uma organização adequada do projeto é essencial para manter a arquitetura limpa. Antes de configurar a estrutura do seu projeto, certifique-se de estar usando ambientes virtuais do Python para isolar dependências. O [venv Cheatsheet](https://www.glukhov.org/pt/post/2025/05/python-venv-cheatsheet/ "venv Cheatsheet") aborda tudo o que você precisa saber sobre gerenciamento de ambientes virtuais. Para projetos modernos do Python, considere usar [uv - New Python Package, Project, and Environment Manager](https://www.glukhov.org/pt/post/2025/06/uv-new-python-package-project-and-environment-manager/ "UV é um novo gerenciador de pacotes, projetos e ambientes do Python."), que oferece gerenciamento de pacotes e configuração de projetos mais rápidos.

```text
my_application/
├── domain/                 # Regras de negócio empresariais
│   ├── __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 de 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 Arquitetura Limpa

### Teste Unitário da Lógica de Domínio

```python
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
`

### Teste de Integração com Repositório

```python
@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
`

### Teste da Camada de Serviço

```python
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

### Superdimensionamento

Não implemente a arquitetura limpa para aplicações simples de CRUD. Comece simples e refatore conforme a complexidade cresce.

### Abstrações Vazias

Certifique-se de que as entidades do domínio não contenham anotações de banco de dados ou código específico de framework:

```python
# Ruim
from sqlalchemy import Column

@dataclass
class User:
    id: Column(Integer, primary_key=True)  # Código do framework infiltrando no domínio

# Bom
@dataclass
class User:
    id: UUID  # Objeto puro do domínio
`

### Dependências Circulares

Use injeção de dependência e interfaces para quebrar dependências circulares entre camadas.

### Ignorar o Contexto

A arquitetura limpa não é uma solução universal. Ajuste a rigidez das camadas com base no tamanho do projeto e na expertise da equipe.

## Links Úteis

- [Arquitetura Limpa por Robert C. Martin](https://blog.cleancoder.com/uncle-bob/2012/08/13/the-clean-architecture.html)
- [Documentação de Dicas de Tipo do Python](https://docs.python.org/3/library/typing.html)
- [Documentação do Pydantic](https://docs.pydantic.dev/)
- [Documentação Oficial do FastAPI](https://fastapi.tiangolo.com/)
- [Documentação ORM do SQLAlchemy](https://docs.sqlalchemy.org/)
- [Biblioteca de Injeção de Dependência](https://python-dependency-injector.ets-labs.org/)
- [Referência de Design Orientado a Domínio](https://www.domainlanguage.com/ddd/reference/)
- [Padrões de Arquitetura com Python](https://www.cosmicpython.com/)
- [Blog de Martin Fowler sobre Arquitetura](https://martinfowler.com/architecture/)
- [Guia de Padrões de Projeto em Python](https://refactoring.guru/design-patterns/python)
- [Python Cheatsheet](https://www.glukhov.org/pt/post/2024/08/python-cheat-sheet/ "Python Cheatsheet")
- [venv Cheatsheet](https://www.glukhov.org/pt/post/2025/05/python-venv-cheatsheet/ "venv Cheatsheet")
- [uv - New Python Package, Project, and Environment Manager](https://www.glukhov.org/pt/post/2025/06/uv-new-python-package-project-and-environment-manager/ "UV é um novo gerenciador de pacotes, projetos e ambientes do Python.")
- [AWS lambda performance: JavaScript vs Python vs Golang](https://www.glukhov.org/pt/post/2024/08/aws-lambda-golang-python-js/ "Desenvolvendo aplicações sem servidor na AWS com sam usando JavaScript, Python e Golang")
- [Construindo Microserviços Event-Driven com AWS Kinesis](https://www.glukhov.org/pt/post/2025/11/service-oriented-and-microservices-with-aws-kinesis/ "Guia abrangente para implementar uma arquitetura de microserviços event-driven escalável usando AWS Kinesis Data Streams, incluindo processamento de dados em tempo real, desacoplamento de serviços e melhores práticas para implantações em produção.")