Python Design Patterns for Clean Architecture

Build maintainable Python apps with SOLID design patterns

Clean Architecture has revolutionized how developers build scalable, maintainable applications by emphasizing separation of concerns and dependency management.

In Python, these principles combine with the language’s dynamic nature to create flexible, testable systems that evolve with business requirements without becoming technical debt.

vibrant tech conference hall

Understanding Clean Architecture in Python

Clean Architecture, introduced by Robert C. Martin (Uncle Bob), organizes software into concentric layers where dependencies point inward toward core business logic. This architectural pattern ensures that your application’s critical business rules remain independent of frameworks, databases, and external services.

The Core Philosophy

The fundamental principle is simple yet powerful: business logic should not depend on infrastructure. Your domain entities, use cases, and business rules should work regardless of whether you’re using PostgreSQL or MongoDB, FastAPI or Flask, AWS or Azure.

In Python, this philosophy aligns perfectly with the language’s “duck typing” and protocol-oriented programming, allowing clean separation without the ceremony required in statically typed languages.

The Four Layers of Clean Architecture

Entities Layer (Domain): Pure business objects with enterprise-wide business rules. These are POJOs (Plain Old Python Objects) with no external dependencies.

Use Cases Layer (Application): Application-specific business rules orchestrating the flow of data between entities and external services.

Interface Adapters Layer: Converts data between the format most convenient for use cases and entities, and the format required by external agencies.

Frameworks & Drivers Layer: All the external details like databases, web frameworks, and external APIs.

SOLID Principles in Python

SOLID principles form the foundation of clean architecture. Let’s explore how each principle manifests in Python. For a comprehensive overview of design patterns in Python, see the Python Design Patterns Guide.

Single Responsibility Principle (SRP)

Each class should have one reason to change:

# Bad: Multiple responsibilities
class UserManager:
    def create_user(self, user_data):
        # Create user
        pass
    
    def send_welcome_email(self, user):
        # Send email
        pass
    
    def log_creation(self, user):
        # Log to file
        pass

# Good: Separated responsibilities
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"User created: {user.id}")
        return user

Open/Closed Principle (OCP)

Software entities should be open for extension but closed for modification:

from abc import ABC, abstractmethod
from typing import Protocol

# Using Protocol (Python 3.8+)
class PaymentProcessor(Protocol):
    def process_payment(self, amount: float) -> bool:
        ...

class CreditCardProcessor:
    def process_payment(self, amount: float) -> bool:
        # Credit card logic
        return True

class PayPalProcessor:
    def process_payment(self, amount: float) -> bool:
        # PayPal logic
        return True

# Easily extensible without modifying existing code
class CryptoProcessor:
    def process_payment(self, amount: float) -> bool:
        # Cryptocurrency logic
        return True

Liskov Substitution Principle (LSP)

Objects should be replaceable with their subtypes without breaking the program:

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:
        # PostgreSQL implementation
        pass
    
    def get(self, key: str) -> str:
        # PostgreSQL implementation
        return ""

class RedisStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # Redis implementation
        pass
    
    def get(self, key: str) -> str:
        # Redis implementation
        return ""

# Both can be used interchangeably
def process_data(store: DataStore, key: str, value: str):
    store.save(key, value)
    return store.get(key)

Interface Segregation Principle (ISP)

Clients shouldn’t be forced to depend on interfaces they don’t use:

# Bad: Fat interface
class Worker(ABC):
    @abstractmethod
    def work(self): pass
    
    @abstractmethod
    def eat(self): pass
    
    @abstractmethod
    def sleep(self): pass

# Good: Segregated interfaces
class Workable(Protocol):
    def work(self) -> None: ...

class Eatable(Protocol):
    def eat(self) -> None: ...

class Human:
    def work(self) -> None:
        print("Working")
    
    def eat(self) -> None:
        print("Eating")

class Robot:
    def work(self) -> None:
        print("Working")
    # No eat method needed

Dependency Inversion Principle (DIP)

High-level modules shouldn’t depend on low-level modules. Both should depend on abstractions:

from typing import Protocol

# Abstraction
class EmailSender(Protocol):
    def send(self, to: str, subject: str, body: str) -> None:
        ...

# Low-level module
class SMTPEmailSender:
    def send(self, to: str, subject: str, body: str) -> None:
        # SMTP implementation
        pass

# High-level module depends on abstraction
class UserRegistrationService:
    def __init__(self, email_sender: EmailSender):
        self.email_sender = email_sender
    
    def register(self, email: str, name: str):
        # Registration logic
        self.email_sender.send(
            to=email,
            subject="Welcome!",
            body=f"Hello {name}"
        )

Repository Pattern: Abstracting Data Access

The Repository Pattern provides a collection-like interface for accessing domain objects, hiding the details of data storage.

Basic Repository Implementation

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

SQLAlchemy Implementation

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

In-Memory Repository for Testing

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

Service Layer: Orchestrating Business Logic

The Service Layer implements use cases and orchestrates the flow between repositories, external services, and domain logic.

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:
        # Check if user exists
        existing_user = self.user_repository.get_by_email(email)
        if existing_user:
            raise UserAlreadyExistsError(f"User with email {email} already exists")
        
        # Create new user
        user = User(
            id=uuid4(),
            email=email,
            name=name,
            is_active=True
        )
        
        # Save to repository
        user = self.user_repository.save(user)
        
        # Send welcome email
        self.email_service.send(
            to=user.email,
            subject="Welcome!",
            body=f"Hello {user.name}, welcome to our platform!"
        )
        
        # Publish event
        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"User {user_id} not found")
        
        user.is_active = False
        user = self.user_repository.save(user)
        
        self.event_publisher.publish('user.deactivated', {
            'user_id': str(user.id)
        })
        
        return user

Dependency Injection in Python

Python’s dynamic nature makes dependency injection straightforward without requiring heavy frameworks.

Constructor Injection

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):
        # Use injected dependencies
        pass

Simple Dependency Container

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"No registration found for {interface}")

# Usage
def create_container() -> Container:
    container = Container()
    
    # Register services
    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

Hexagonal Architecture (Ports and Adapters)

Hexagonal Architecture places business logic at the center with adapters handling external communication.

Defining Ports (Interfaces)

# Input Port (Primary)
class CreateUserUseCase(Protocol):
    def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
        ...

# Output Port (Secondary)
class UserPersistencePort(Protocol):
    def save(self, user: User) -> User:
        ...
    
    def find_by_email(self, email: str) -> Optional[User]:
        ...

Implementing 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 (Database)
# Already implemented as SQLAlchemyUserRepository

Domain-Driven Design Patterns

Value Objects

Immutable objects defined by their attributes:

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"Invalid email: {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("Amount cannot be negative")
        if self.currency not in ['USD', 'EUR', 'GBP']:
            raise ValueError(f"Unsupported currency: {self.currency}")
    
    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)

Aggregates

Cluster of domain objects treated as a single unit:

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("Cannot confirm empty order")
        if self.status != "pending":
            raise ValueError("Order already processed")
        self.status = "confirmed"

Domain Events

Domain events enable loose coupling between components and support event-driven architectures. For production-scale event-driven systems, consider implementing event streaming with services like AWS Kinesis—see Building Event-Driven Microservices with AWS Kinesis for a detailed guide.

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)

Modern Python Features for Clean Architecture

Python’s modern features make implementing clean architecture more elegant and type-safe. If you need a quick reference for Python syntax and features, check out the Python Cheatsheet.

Type Hints and 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 for Validation

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  # Make immutable

Async/Await for I/O Operations

Python’s async/await syntax is particularly powerful for I/O-bound operations in clean architecture, allowing non-blocking interactions with databases and external services. When deploying Python applications to serverless platforms, understanding performance characteristics becomes crucial—see AWS lambda performance: JavaScript vs Python vs Golang for insights on optimizing Python serverless functions.

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]

Project Structure Best Practices

Proper project organization is essential for maintaining clean architecture. Before setting up your project structure, ensure you’re using Python virtual environments for dependency isolation. The venv Cheatsheet covers everything you need to know about managing virtual environments. For modern Python projects, consider using uv - New Python Package, Project, and Environment Manager, which provides faster package management and project setup.

my_application/
├── domain/                 # Enterprise business rules
│   ├── __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/            # Application business rules
│   ├── __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/         # External interfaces
│   ├── __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/           # UI/API layer
│   ├── __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                 # Application entry point
├── container.py            # Dependency injection setup
├── pyproject.toml
└── README.md

Testing Clean Architecture

Unit Testing Domain Logic

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

Integration Testing with 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

Testing Service Layer

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()

Common Pitfalls and How to Avoid Them

Over-Engineering

Don’t implement clean architecture for simple CRUD applications. Start simple and refactor as complexity grows.

Leaky Abstractions

Ensure domain entities don’t contain database annotations or framework-specific code:

# Bad
from sqlalchemy import Column

@dataclass
class User:
    id: Column(Integer, primary_key=True)  # Framework leaking into domain

# Good
@dataclass
class User:
    id: UUID  # Pure domain object

Circular Dependencies

Use dependency injection and interfaces to break circular dependencies between layers.

Ignoring Context

Clean Architecture isn’t one-size-fits-all. Adjust layer strictness based on project size and team expertise.