Pular para conteúdo

Design Patterns

Padrões de design de código utilizados no projeto.

Backend Patterns (Python)

Repository Pattern

Abstrai acesso ao banco de dados.

from abc import ABC, abstractmethod
from typing import Generic, TypeVar, Optional, List

T = TypeVar('T')

class Repository(Generic[T], ABC):
    """Base repository interface."""

    @abstractmethod
    async def get(self, id: int) -> Optional[T]:
        """Get entity by ID."""
        pass

    @abstractmethod
    async def list(self, skip: int = 0, limit: int = 100) -> List[T]:
        """List entities with pagination."""
        pass

    @abstractmethod
    async def create(self, entity: T) -> T:
        """Create new entity."""
        pass

    @abstractmethod
    async def update(self, id: int, entity: T) -> Optional[T]:
        """Update existing entity."""
        pass

    @abstractmethod
    async def delete(self, id: int) -> bool:
        """Delete entity."""
        pass


class UserRepository(Repository[User]):
    """Concrete implementation for User."""

    def __init__(self, session: AsyncSession):
        self.session = session

    async def get(self, id: int) -> Optional[User]:
        result = await self.session.execute(
            select(User).where(User.id == id)
        )
        return result.scalar_one_or_none()

    async def list(self, skip: int = 0, limit: int = 100) -> List[User]:
        result = await self.session.execute(
            select(User).offset(skip).limit(limit)
        )
        return result.scalars().all()

    # Implement other methods...

Service Layer Pattern

Encapsula business logic.

class UserService:
    """Service layer for user operations."""

    def __init__(self, repository: UserRepository):
        self.repository = repository

    async def create_user(self, request: CreateUserRequest) -> User:
        """Create user with validation and business logic."""

        # Validation
        if await self.repository.get_by_email(request.email):
            raise EmailAlreadyExistsError()

        # Business logic
        user = User(
            email=request.email.lower(),
            name=request.name,
            password_hash=hash_password(request.password)
        )

        # Save
        user = await self.repository.create(user)

        # Side effects (async events)
        await self.event_publisher.publish_user_created(user)

        return user

Dependency Injection

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

async def get_session() -> AsyncSession:
    """Database session dependency."""
    async with async_session_maker() as session:
        yield session

def get_user_repository(
    session: AsyncSession = Depends(get_session)
) -> UserRepository:
    """User repository dependency."""
    return UserRepository(session)

def get_user_service(
    repository: UserRepository = Depends(get_user_repository)
) -> UserService:
    """User service dependency."""
    return UserService(repository)

# Uso no endpoint
@router.post("/users")
async def create_user(
    request: CreateUserRequest,
    service: UserService = Depends(get_user_service)
):
    return await service.create_user(request)

Factory Pattern

class MessageHandlerFactory:
    """Factory for creating message handlers."""

    _handlers = {}

    @classmethod
    def register(cls, event_type: str, handler):
        """Register handler for event type."""
        cls._handlers[event_type] = handler

    @classmethod
    def get_handler(cls, event_type: str):
        """Get handler for event type."""
        handler = cls._handlers.get(event_type)
        if not handler:
            raise ValueError(f"No handler for {event_type}")
        return handler

# Register handlers
MessageHandlerFactory.register('user_created', handle_user_created)
MessageHandlerFactory.register('user_updated', handle_user_updated)

# Use in Lambda
def lambda_handler(event, context):
    for record in event['Records']:
        message = json.loads(record['body'])
        handler = MessageHandlerFactory.get_handler(message['event_type'])
        handler(message)

Singleton Pattern

class DatabaseConnection:
    """Singleton database connection."""

    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._engine = create_async_engine(DATABASE_URL)
        return cls._instance

    @property
    def engine(self):
        return self._engine

# Usage
db = DatabaseConnection()
engine = db.engine

Frontend Patterns (React)

Container/Presentational Pattern

// Presentational Component (dumb)
interface UserListProps {
  users: User[];
  onUserClick: (user: User) => void;
  loading: boolean;
}

export function UserList({ users, onUserClick, loading }: UserListProps) {
  if (loading) return <Loader />;

  return (
    <ul>
      {users.map(user => (
        <li key={user.id} onClick={() => onUserClick(user)}>
          {user.name}
        </li>
      ))}
    </ul>
  );
}

// Container Component (smart)
export function UserListContainer() {
  const { users, loading } = useUsers();
  const navigate = useNavigate();

  const handleUserClick = (user: User) => {
    navigate(`/users/${user.id}`);
  };

  return (
    <UserList 
      users={users}
      onUserClick={handleUserClick}
      loading={loading}
    />
  );
}

Custom Hooks Pattern

// Encapsular lógica reutilizável
export function useApi<T>(endpoint: string) {
  const [data, setData] = useState<T | null>(null);
  const [loading, setLoading] = useState(true);
  const [error, setError] = useState<Error | null>(null);

  const fetch = useCallback(async () => {
    try {
      setLoading(true);
      const response = await apiClient.get<T>(endpoint);
      setData(response.data);
    } catch (err) {
      setError(err as Error);
    } finally {
      setLoading(false);
    }
  }, [endpoint]);

  useEffect(() => {
    fetch();
  }, [fetch]);

  return { data, loading, error, refetch: fetch };
}

// Uso
function UserProfile({ userId }: Props) {
  const { data: user, loading } = useApi<User>(`/users/${userId}`);
  // ...
}

Compound Components Pattern

// Componentes que trabalham juntos
function Card({ children }: PropsWithChildren) {
  return <div className="card">{children}</div>;
}

Card.Header = function CardHeader({ children }: PropsWithChildren) {
  return <div className="card-header">{children}</div>;
};

Card.Body = function CardBody({ children }: PropsWithChildren) {
  return <div className="card-body">{children}</div>;
};

Card.Footer = function CardFooter({ children }: PropsWithChildren) {
  return <div className="card-footer">{children}</div>;
};

// Uso
<Card>
  <Card.Header>Title</Card.Header>
  <Card.Body>Content</Card.Body>
  <Card.Footer>Actions</Card.Footer>
</Card>

Render Props Pattern

interface DataProviderProps<T> {
  url: string;
  children: (data: T, loading: boolean, error: Error | null) => React.ReactNode;
}

function DataProvider<T>({ url, children }: DataProviderProps<T>) {
  const { data, loading, error } = useApi<T>(url);
  return <>{children(data, loading, error)}</>;
}

// Uso
<DataProvider<User[]> url="/api/users">
  {(users, loading, error) => {
    if (loading) return <Loader />;
    if (error) return <Error message={error.message} />;
    return <UserList users={users} />;
  }}
</DataProvider>

General Patterns

Strategy Pattern

from abc import ABC, abstractmethod

class PaymentStrategy(ABC):
    @abstractmethod
    async def process(self, amount: float, details: dict) -> PaymentResult:
        pass

class CreditCardPayment(PaymentStrategy):
    async def process(self, amount: float, details: dict) -> PaymentResult:
        # Process credit card
        ...

class PIXPayment(PaymentStrategy):
    async def process(self, amount: float, details: dict) -> PaymentResult:
        # Process PIX
        ...

class PaymentService:
    def __init__(self):
        self.strategies = {
            'credit_card': CreditCardPayment(),
            'pix': PIXPayment(),
        }

    async def process_payment(
        self, 
        method: str, 
        amount: float, 
        details: dict
    ) -> PaymentResult:
        strategy = self.strategies.get(method)
        if not strategy:
            raise ValueError(f"Unknown payment method: {method}")
        return await strategy.process(amount, details)

Observer Pattern (Event-driven)

from typing import List, Callable

class EventEmitter:
    """Simple event emitter."""

    def __init__(self):
        self._listeners: dict[str, List[Callable]] = {}

    def on(self, event: str, callback: Callable):
        """Subscribe to event."""
        if event not in self._listeners:
            self._listeners[event] = []
        self._listeners[event].append(callback)

    async def emit(self, event: str, data: dict):
        """Emit event to all listeners."""
        if event not in self._listeners:
            return

        for callback in self._listeners[event]:
            await callback(data)

# Usage
emitter = EventEmitter()

emitter.on('user_created', lambda data: send_welcome_email(data))
emitter.on('user_created', lambda data: track_analytics(data))
emitter.on('user_created', lambda data: notify_admins(data))

await emitter.emit('user_created', {'user_id': 123})

Circuit Breaker Pattern

Para chamadas a serviços externos:

from enum import Enum
import time

class CircuitState(Enum):
    CLOSED = "closed"  # Normal operation
    OPEN = "open"      # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if recovered

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = CircuitState.CLOSED

    async def call(self, func, *args, **kwargs):
        """Execute function with circuit breaker protection."""

        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise CircuitBreakerOpenError("Service unavailable")

        try:
            result = await func(*args, **kwargs)
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise

    def on_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED

    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

# Usage
breaker = CircuitBreaker()

async def call_external_api():
    return await breaker.call(external_api.get_data)

Retry Pattern

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def fetch_data_with_retry(url: str):
    """Fetch data with exponential backoff retry."""
    response = await http_client.get(url)
    response.raise_for_status()
    return response.json()

Anti-Patterns (Evitar)

❌ God Object

# Ruim: Classe faz tudo
class UserManager:
    def create_user(self): ...
    def send_email(self): ...
    def process_payment(self): ...
    def generate_report(self): ...
    def update_cache(self): ...

❌ Spaghetti Code

# Ruim: Lógica entrelaçada
def process_order(order):
    if order.status == 'pending':
        if order.payment:
            if order.payment.verified:
                # 50 linhas de lógica aqui...
                # E mais 50 linhas...

❌ Copy-Paste Code

# Ruim: Duplicação
def send_welcome_email(user):
    # 20 linhas de código

def send_password_reset_email(user):
    # 20 linhas MUITO similares

# Bom: Extrair comum
def send_templated_email(user, template, context):
    # Lógica comum aqui

def send_welcome_email(user):
    send_templated_email(user, 'welcome', {'name': user.name})

Referências