The Problem Clean Architecture Solves
You've got a Flask app that started simple. Routes call service functions, service functions query the database directly, and business logic is scattered across all three layers. Six months in, you want to swap PostgreSQL for DynamoDB and realize the database-specific code is woven into everything. Or you need to add a CLI interface alongside the HTTP API, and your "services" are tightly coupled to Flask's request context.
Clean architecture addresses this by enforcing a dependency rule: inner layers (business logic) don't know about outer layers (frameworks, databases, HTTP). The business rules don't care whether data comes from a REST API or a CLI command, or whether it's stored in PostgreSQL or a CSV file.
Three Flavors, One Idea
Hexagonal Architecture (Alistair Cockburn, 2005), Onion Architecture (Jeffrey Palermo, 2008), and Clean Architecture (Robert Martin, 2012) are all expressing the same core concept with different terminology. The differences are mostly academic.
The Common Structure
At the center: Domain entities and business rules. These are pure objects with no dependencies on frameworks or infrastructure. An Order entity knows that it can't have a negative total and that it transitions through specific states. It doesn't know about databases or HTTP.
Next ring: Use cases / Application services. These orchestrate business rules. A PlaceOrder use case coordinates creating the order, checking inventory, and triggering payment. It depends on the domain layer and on abstract interfaces (ports) for infrastructure.
Outer ring: Adapters / Infrastructure. Database repositories, HTTP controllers, message queue consumers, email services. These implement the abstract interfaces defined by the inner layers.
# Domain entity — no imports from frameworks or infrastructure
class Order:
def __init__(self, customer_id: str, items: list[OrderItem]):
if not items:
raise ValueError("Order must have at least one item")
self.customer_id = customer_id
self.items = items
self.status = OrderStatus.PENDING
self.total = sum(item.price * item.quantity for item in items)
def confirm(self):
if self.status != OrderStatus.PENDING:
raise InvalidStateTransition(
f"Can't confirm order in {self.status} state"
)
self.status = OrderStatus.CONFIRMED
def cancel(self):
if self.status == OrderStatus.SHIPPED:
raise InvalidStateTransition("Can't cancel shipped order")
self.status = OrderStatus.CANCELLEDPorts and Adapters (Hexagonal)
In hexagonal architecture, "ports" are interfaces that the application core defines. "Adapters" are implementations that connect the core to the outside world.
Driving ports (left side) are how the outside world triggers application logic: an HTTP controller, a CLI command handler, a message consumer. They call into the application core.
Driven ports (right side) are how the application core accesses infrastructure: a repository interface, a payment gateway interface, a notification service interface. The core defines the interface; the infrastructure provides the implementation.
# Port (defined by the application core)
from abc import ABC, abstractmethod
class OrderRepository(ABC):
@abstractmethod
def save(self, order: Order) -> None: ...
@abstractmethod
def find_by_id(self, order_id: str) -> Order | None: ...
@abstractmethod
def find_by_customer(self, customer_id: str) -> list[Order]: ...
# Adapter (infrastructure implementation)
class PostgresOrderRepository(OrderRepository):
def __init__(self, connection_pool):
self.pool = connection_pool
def save(self, order: Order) -> None:
with self.pool.connection() as conn:
conn.execute(
"INSERT INTO orders (id, customer_id, status, total) "
"VALUES (%s, %s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET status=%s, total=%s",
(order.id, order.customer_id, order.status.value,
order.total, order.status.value, order.total)
)
def find_by_id(self, order_id: str) -> Order | None:
with self.pool.connection() as conn:
row = conn.execute(
"SELECT * FROM orders WHERE id = %s", (order_id,)
).fetchone()
return self._to_entity(row) if row else NoneUse Cases as the Application's API
Use cases (or application services) are where your business workflows live. They receive a command, orchestrate domain entities and driven ports, and return a result.
class PlaceOrderUseCase:
def __init__(
self,
order_repo: OrderRepository,
inventory: InventoryService,
payment: PaymentGateway,
notifier: NotificationService
):
self.order_repo = order_repo
self.inventory = inventory
self.payment = payment
self.notifier = notifier
def execute(self, command: PlaceOrderCommand) -> OrderResult:
# Build domain entity
items = [OrderItem(i.sku, i.quantity, i.price)
for i in command.items]
order = Order(command.customer_id, items)
# Check inventory
for item in items:
if not self.inventory.check_available(item.sku, item.quantity):
raise InsufficientInventory(item.sku)
# Reserve inventory
self.inventory.reserve(order.id, items)
try:
# Process payment
self.payment.charge(command.customer_id, order.total)
order.confirm()
except PaymentFailed:
self.inventory.release(order.id, items)
order.cancel()
raise
self.order_repo.save(order)
self.notifier.send_order_confirmation(order)
return OrderResult(order_id=order.id, status=order.status)Notice that the use case depends only on abstractions (interfaces). It doesn't import Flask, SQLAlchemy, or Stripe. You could test it with in-memory fakes for every dependency.
When Is This Worth the Complexity?
Here's where I'll push back on clean architecture evangelists: this pattern adds real complexity. More files, more abstractions, more indirection. You need to weigh that against the benefits.
Worth it when:
- The application has complex business logic that changes independently from infrastructure concerns
- You expect to swap infrastructure components (different database, different message broker)
- Multiple interfaces exist (HTTP API + CLI + event consumer all triggering the same workflows)
- The team is large enough that clear boundaries prevent stepping on each other's code
- Thorough unit testing of business logic is important (testing without database setup)
Not worth it when:
- It's a CRUD app with minimal business logic — the "clean" architecture just adds ceremony around pass-through database calls
- You're building a prototype or MVP that might be thrown away
- The team is small (2-3 people) and everyone understands the whole codebase
- You have one interface and one database and no plans to change either
The worst outcome is applying clean architecture to a CRUD service and ending up with a UserRepository interface that has exactly one implementation and a CreateUserUseCase that does nothing but call repo.save(user). That's not architecture — it's bureaucracy.
Project Structure
Here's a directory layout that works well for medium-sized Python projects:
src/
domain/
entities/ # Order, Customer, Product
value_objects/ # Money, Address, Email
exceptions.py # Domain-specific errors
application/
use_cases/ # PlaceOrder, CancelOrder
ports/ # Repository interfaces, service interfaces
dto.py # Command/result objects
infrastructure/
persistence/ # PostgresOrderRepository, RedisCache
http/ # Flask routes, FastAPI endpoints
messaging/ # Kafka consumer, RabbitMQ publisher
external/ # Stripe adapter, SendGrid adapter
config.py # Dependency injection / wiringThe key rule: arrows point inward. infrastructure imports from application and domain. application imports from domain. domain imports from nothing outside itself.
Dependency injection wires everything together at the entry point. In Python, you can use a simple factory function or a DI container like dependency-injector. Don't overthink this part — constructor injection and a composition root function work fine for most projects.