Clean Architecture: Layered Architecture in Practice

# Clean Architecture Dependency Rule
# Inner layers should know nothing about outer layers

dependency_rule = """
GOOD:  Domain ← Application ← Infrastructure ← Presentation
       (Domain has no dependencies)

BAD:   Presentation → Domain
       (UI directly imports Domain - VIOLATES RULE)

SOLUTION: Use interfaces/abstract classes in inner layers
          Implement in outer layers (Dependency Inversion)
"""

# domain/entities.py

from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import uuid

class Order:
    """Order entity - core business logic"""
    
    def __init__(self, customer_id: str, items: list[OrderItem]):
        self.id = str(uuid.uuid4())
        self.customer_id = customer_id
        self.items = items
        self.status = OrderStatus.PENDING
        self.created_at = datetime.utcnow()
        self.updated_at = datetime.utcnow()
    
    @property
    def total(self) -> float:
        return sum(item.price * item.quantity for item in self.items)
    
    def can_cancel(self) -> bool:
        return self.status in [OrderStatus.PENDING, OrderStatus.CONFIRMED]
    
    def cancel(self):
        if not self.can_cancel():
            raise ValueError(f"Cannot cancel order in status {self.status}")
        self.status = OrderStatus.CANCELLED
        self.updated_at = datetime.utcnow()
    
    def complete(self):
        if self.status != OrderStatus.SHIPPED:
            raise ValueError("Only shipped orders can be completed")
        self.status = OrderStatus.COMPLETED
        self.updated_at = datetime.utcnow()


@dataclass(frozen=True)
class Money:
    """Value object - immutable"""
    amount: float
    currency: str
    
    def __add__(self, other: 'Money') -> 'Money':
        if self.currency != other.currency:
            raise ValueError("Cannot add different currencies")
        return Money(self.amount + other.amount, self.currency)


class OrderStatus:
    PENDING = "pending"
    CONFIRMED = "confirmed"
    SHIPPED = "shipped"
    COMPLETED = "completed"
    CANCELLED = "cancelled"


@dataclass
class OrderItem:
    product_id: str
    name: str
    quantity: int
    price: float

# domain/events.py

from dataclasses import dataclass
from datetime import datetime
from typing import Any
import uuid

@dataclass
class DomainEvent:
    """Base class for domain events"""
    event_id: str
    occurred_on: datetime
    
    def __init__(self):
        self.event_id = str(uuid.uuid4())
        self.occurred_on = datetime.utcnow()


class OrderCreatedEvent(DomainEvent):
    def __init__(self, order_id: str, customer_id: str, total: float):
        super().__init__()
        self.order_id = order_id
        self.customer_id = customer_id
        self.total = total


class OrderCancelledEvent(DomainEvent):
    def __init__(self, order_id: str, reason: str):
        super().__init__()
        self.order_id = order_id
        self.reason = reason


class OrderShippedEvent(DomainEvent):
    def __init__(self, order_id: str, tracking_number: str):
        super().__init__()
        self.order_id = order_id
        self.tracking_number = tracking_number

# application/use_cases.py

from abc import ABC, abstractmethod
from typing import List
from domain.entities import Order, OrderItem
from domain.events import DomainEvent

# Ports (Interfaces) - Defined in application layer
class OrderRepository(ABC):
    """Port for order persistence"""
    
    @abstractmethod
    def save(self, order: Order) -> Order:
        pass
    
    @abstractmethod
    def find_by_id(self, order_id: str) -> Order:
        pass
    
    @abstractmethod
    def find_by_customer(self, customer_id: str) -> List[Order]:
        pass


class EventPublisher(ABC):
    """Port for publishing domain events"""
    
    @abstractmethod
    def publish(self, event: DomainEvent):
        pass


# Use Cases (Application Services)
class CreateOrderUseCase:
    def __init__(self, order_repo: OrderRepository, 
                 event_publisher: EventPublisher):
        self.order_repo = order_repo
        self.event_publisher = event_publisher
    
    def execute(self, command: CreateOrderCommand) -> Order:
        # Create order entity
        order = Order(
            customer_id=command.customer_id,
            items=[OrderItem(**item) for item in command.items]
        )
        
        # Persist
        saved_order = self.order_repo.save(order)
        
        # Publish event
        from domain.events import OrderCreatedEvent
        event = OrderCreatedEvent(
            order_id=saved_order.id,
            customer_id=saved_order.customer_id,
            total=saved_order.total
        )
        self.event_publisher.publish(event)
        
        return saved_order


class CancelOrderUseCase:
    def __init__(self, order_repo: OrderRepository,
                 event_publisher: EventPublisher):
        self.order_repo = order_repo
        self.event_publisher = event_publisher
    
    def execute(self, command: CancelOrderCommand) -> Order:
        # Load order
        order = self.order_repo.find_by_id(command.order_id)
        
        # Domain logic
        order.cancel()
        
        # Persist
        saved_order = self.order_repo.save(order)
        
        # Publish event
        event = OrderCancelledEvent(
            order_id=command.order_id,
            reason=command.reason
        )
        self.event_publisher.publish(event)
        
        return saved_order


# DTOs
@dataclass
class CreateOrderCommand:
    customer_id: str
    items: List[dict]


@dataclass
class CancelOrderCommand:
    order_id: str
    reason: str

# infrastructure/repositories.py

import sqlite3
from typing import List
from domain.entities import Order, OrderItem, OrderStatus
from application.use_cases import OrderRepository


class SQLiteOrderRepository(OrderRepository):
    """Implement OrderRepository using SQLite"""
    
    def __init__(self, connection):
        self.connection = connection
    
    def save(self, order: Order) -> Order:
        cursor = self.connection.cursor()
        
        # Insert order
        cursor.execute("""
            INSERT INTO orders (id, customer_id, status, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?)
        """, [order.id, order.customer_id, order.status, 
              order.created_at, order.updated_at])
        
        # Insert items
        for item in order.items:
            cursor.execute("""
                INSERT INTO order_items (id, order_id, product_id, name, quantity, price)
                VALUES (?, ?, ?, ?, ?, ?)
            """, [f"{order.id}-{item.product_id}", order.id, 
                  item.product_id, item.name, item.quantity, item.price])
        
        self.connection.commit()
        return order
    
    def find_by_id(self, order_id: str) -> Order:
        cursor = self.connection.cursor()
        
        cursor.execute("SELECT * FROM orders WHERE id = ?", [order_id])
        row = cursor.fetchone()
        
        if not row:
            return None
        
        # Load items
        cursor.execute("SELECT * FROM order_items WHERE order_id = ?", [order_id])
        item_rows = cursor.fetchall()
        
        items = [OrderItem(
            product_id=r[2], name=r[3], quantity=r[4], price=r[5]
        ) for r in item_rows]
        
        order = Order.__new__(Order)
        order.id = row[0]
        order.customer_id = row[1]
        order.status = row[2]
        order.items = items
        return order
    
    def find_by_customer(self, customer_id: str) -> List[Order]:
        # Similar implementation
        pass


# infrastructure/event_publisher.py

import json
from kafka import KafkaProducer
from domain.events import DomainEvent


class KafkaEventPublisher(EventPublisher):
    def __init__(self, bootstrap_servers):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v.__dict__).encode('utf-8')
        )
    
    def publish(self, event: DomainEvent):
        topic = event.__class__.__name__.replace('Event', '').lower()
        self.producer.send(topic, event)

# presentation/api.py

from flask import Flask, request, jsonify
from application.use_cases import CreateOrderUseCase, CreateOrderCommand
from infrastructure.repositories import SQLiteOrderRepository
from infrastructure.event_publisher import KafkaEventPublisher

app = Flask(__name__)

# Dependency injection (manually or use DI container)
def get_order_use_case():
    conn = sqlite3.connect('orders.db')
    repo = SQLiteOrderRepository(conn)
    publisher = KafkaEventPublisher(['localhost:9092'])
    return CreateOrderUseCase(repo, publisher)


@app.route('/orders', methods=['POST'])
def create_order():
    data = request.json
    
    command = CreateOrderCommand(
        customer_id=data['customer_id'],
        items=data['items']
    )
    
    use_case = get_order_use_case()
    order = use_case.execute(command)
    
    return jsonify({
        'id': order.id,
        'status': order.status,
        'total': order.total
    }), 201


if __name__ == '__main__':
    app.run()

# Using a DI container for cleaner code

from dependency_injector import containers, providers

class Container(containers.DeclarativeContainer):
    # Configuration
    config = providers.Configuration()
    
    # Infrastructure
    database = providers.Singleton(
        sqlite3.connect,
        config.database.path
    )
    
    # Repositories
    order_repository = providers.Factory(
        SQLiteOrderRepository,
        connection=database
    )
    
    # Event publishers
    event_publisher = providers.Factory(
        KafkaEventPublisher,
        bootstrap_servers=config.kafka.servers
    )
    
    # Use cases
    create_order_use_case = providers.Factory(
        CreateOrderUseCase,
        order_repo=order_repository,
        event_publisher=event_publisher
    )


# Usage
container = Container(config={...})
use_case = container.create_order_use_case()
order = use_case.execute(command)

# Clean Architecture Directory Structure

src/
├── domain/                    # Innermost layer - no dependencies
│   ├── entities/
│   │   └── order.py
│   ├── value_objects/
│   │   └── money.py
│   ├── events/
│   │   └── order_events.py
│   ├── repositories/          # Repository interfaces
│   │   └── interfaces.py
│   └── services/              # Domain services
│       └── pricing.py
│
├── application/               # Use cases
│   ├── use_cases/
│   │   ├── create_order.py
│   │   └── cancel_order.py
│   ├── dto/
│   │   └── order_dto.py
│   ├── ports/                 # Port interfaces
│   │   ├── order_repository.py
│   │   └── event_publisher.py
│   └── services/             # Application services
│       └── order_service.py
│
├── infrastructure/            # Outer layer - implements ports
│   ├── repositories/
│   │   └── sqlite_order_repository.py
│   ├── event_publishers/
│   │   └── kafka_publisher.py
│   ├── external_services/
│   │   └── payment_gateway.py
│   └── database/
│       └── sqlite_connection.py
│
└── presentation/              # Outer layer - API, CLI, etc
    ├── api/
    │   ├── flask_app.py
    │   └── controllers/
    ├── cli/
    │   └── commands.py
    └── views/
        └── templates/

# Testing structure

# tests/unit/domain/test_order.py

import pytest
from domain.entities import Order, OrderItem, OrderStatus

def test_order_total_calculation():
    items = [
        OrderItem(product_id="p1", name="Widget", quantity=2, price=10.00),
        OrderItem(product_id="p2", name="Gadget", quantity=1, price=25.00)
    ]
    
    order = Order(customer_id="c1", items=items)
    
    assert order.total == 45.00  # 2*10 + 1*25

def test_order_can_cancel_when_pending():
    order = Order(customer_id="c1", items=[])
    
    assert order.can_cancel() == True
    order.cancel()
    assert order.status == OrderStatus.CANCELLED


# tests/integration/test_create_order.py

def test_create_order_use_case():
    # Mock repository
    mock_repo = MockOrderRepository()
    mock_publisher = MockEventPublisher()
    
    use_case = CreateOrderUseCase(mock_repo, mock_publisher)
    
    command = CreateOrderCommand(
        customer_id="c1",
        items=[{"product_id": "p1", "name": "Widget", 
                "quantity": 2, "price": 10.00}]
    )
    
    order = use_case.execute(command)
    
    assert order.id is not None
    assert order.status == OrderStatus.PENDING
    mock_repo.save.assert_called_once()
    mock_publisher.publish.assert_called_once()

advantages:
  - Testability: Easy to test each layer in isolation
  - Maintainability: Clear separation of concerns
  - Flexibility: Swap implementations without changing domain
  - Framework Independence: Domain not tied to frameworks
  - Parallel Development: Teams can work on different layers

disadvantages:
  - Initial Complexity: More boilerplate code
  - Learning Curve: Team needs to understand the pattern
  - Overhead: May be overkill for simple applications
  - Performance: Extra layers may add minimal overhead