Architecture Patterns

# domain/entities/user.pyfrom dataclasses import dataclassfrom datetime import datetime
@dataclassclass User:    """Core user entity - no framework dependencies."""    id: str    email: str    name: str    created_at: datetime    is_active: bool = True
    def deactivate(self):        """Business rule: deactivating user."""        self.is_active = False
    def can_place_order(self) -> bool:        """Business rule: active users can order."""        return self.is_active
# domain/interfaces/user_repository.pyfrom abc import ABC, abstractmethodfrom typing import Optional
class IUserRepository(ABC):    """Port: defines contract, no implementation."""
    @abstractmethod    async def find_by_id(self, user_id: str) -> Optional[User]:        pass
    @abstractmethod    async def save(self, user: User) -> User:        pass
# use_cases/create_user.pyfrom dataclasses import dataclassimport uuidfrom datetime import datetime
@dataclassclass CreateUserRequest:    email: str    name: str
class CreateUserUseCase:    """Use case: orchestrates business logic."""
    def __init__(self, user_repository: IUserRepository):        self.user_repository = user_repository
    async def execute(self, request: CreateUserRequest) -> CreateUserResponse:        # Business validation        existing = await self.user_repository.find_by_email(request.email)        if existing:            return CreateUserResponse(user=None, success=False, error="Email already exists")
        # Create entity        user = User(            id=str(uuid.uuid4()),            email=request.email,            name=request.name,            created_at=datetime.now(),            is_active=True        )
        # Persist        saved_user = await self.user_repository.save(user)        return CreateUserResponse(user=saved_user, success=True)
# adapters/repositories/postgres_user_repository.pyclass PostgresUserRepository(IUserRepository):    """Adapter: PostgreSQL implementation."""
    def __init__(self, pool):        self.pool = pool
    async def find_by_id(self, user_id: str) -> Optional[User]:        async with self.pool.acquire() as conn:            row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)            return self._to_entity(row) if row else None
    async def save(self, user: User) -> User:        async with self.pool.acquire() as conn:            await conn.execute(                """INSERT INTO users (id, email, name, created_at, is_active)                VALUES ($1, $2, $3, $4, $5)                ON CONFLICT (id) DO UPDATE SET email = $2, name = $3, is_active = $5""",                user.id, user.email, user.name, user.created_at, user.is_active            )            return user

# Core domain (hexagon center)class OrderService:    """Domain service - no infrastructure dependencies."""
    def __init__(        self,        order_repository: OrderRepositoryPort,        payment_gateway: PaymentGatewayPort,        notification_service: NotificationPort    ):        self.orders = order_repository        self.payments = payment_gateway        self.notifications = notification_service
    async def place_order(self, order: Order) -> OrderResult:        # Business logic        if not order.is_valid():            return OrderResult(success=False, error="Invalid order")
        # Use ports (interfaces)        payment = await self.payments.charge(amount=order.total, customer=order.customer_id)
        if not payment.success:            return OrderResult(success=False, error="Payment failed")
        order.mark_as_paid()        saved_order = await self.orders.save(order)
        await self.notifications.send(            to=order.customer_email,            subject="Order confirmed",            body=f"Order {order.id} confirmed"        )
        return OrderResult(success=True, order=saved_order)
# Adaptersclass StripePaymentAdapter(PaymentGatewayPort):    """Primary adapter: connects to Stripe API."""    async def charge(self, amount: Money, customer: str) -> PaymentResult:        # Implementation
class MockPaymentAdapter(PaymentGatewayPort):    """Test adapter: no external dependencies."""    async def charge(self, amount: Money, customer: str) -> PaymentResult:        return PaymentResult(success=True, transaction_id="mock-123")

# Value Objects (immutable)@dataclass(frozen=True)class Email:    value: str
    def __post_init__(self):        if "@" not in self.value:            raise ValueError("Invalid email")
@dataclass(frozen=True)class Money:    amount: int  # cents    currency: str
    def add(self, other: "Money") -> "Money":        if self.currency != other.currency:            raise ValueError("Currency mismatch")        return Money(self.amount + other.amount, self.currency)
# Entities (with identity)class Order:    def __init__(self, id: str, customer: Customer):        self.id = id        self.customer = customer        self.items: List[OrderItem] = []        self.status = OrderStatus.PENDING        self._events: List[DomainEvent] = []
    def add_item(self, product: Product, quantity: int):        item = OrderItem(product, quantity)        self.items.append(item)        self._events.append(ItemAddedEvent(self.id, item))
    def submit(self):        if not self.items:            raise ValueError("Cannot submit empty order")        self.status = OrderStatus.SUBMITTED        self._events.append(OrderSubmittedEvent(self.id))
# Aggregates (consistency boundary)class Customer:    def __init__(self, id: str, email: Email):        self.id = id        self.email = email        self._addresses: List[Address] = []
    def add_address(self, address: Address):        if len(self._addresses) >= 5:            raise ValueError("Maximum 5 addresses allowed")        self._addresses.append(address)