Decoupling Your Domain: Refactoring FastAPI with Ports and Adapters

Many developers start a

project by writing business logic directly inside their route handlers. It starts simple. You take a database connection from a dependency, run a few

SQLAlchemy

queries, check some conditions, and return a dictionary. But this approach creates a tangled mess where your core business rules are inseparable from the transport layer and the database implementation.

When your domain logic raises HTTPException, it becomes untestable without a running web server. When it accepts a database Session object, you can't run a unit test without mocking complex

internals or spinning up a real database. This coupling makes switching frameworks—like moving from SQL to NoSQL or

FastAPI

to another web framework—nearly impossible without a total rewrite. The solution is the

Ports and Adapters

architecture, also known as

Hexagonal Architecture

.

Prerequisites

To follow this tutorial, you should be comfortable with

3.10+ and have a baseline understanding of asynchronous programming. You should also understand basic

REST API

concepts and

ORM

. Knowledge of type hinting and

Pydantic

will help you grasp the data modeling sections.

Key Libraries & Tools

• FastAPI
  : A modern, fast web framework for building APIs with Python.
• SQLAlchemy
  : The Python SQL toolkit and ORM used here to manage database interactions.
• typing.Protocol
  : A structural typing mechanism used to define "Ports" without forced inheritance.
• dataclasses
  : Built-in Python decorators used to create clean domain models.

Code Walkthrough: Isolating the Domain

The first step is creating a pure domain layer. This layer must have zero imports from external frameworks. We define our own exceptions to replace HTTP errors.

# domain/errors.py
class DomainError(Exception): 
    pass

class OutOfStock(DomainError):
    def __init__(self, sku: str, requested: int, available: int):
        super().__init__(f"SKU {sku} is out of stock. Requested {requested}, only {available} left.")

Next, we define the "Port." In

, a Port is best represented by a Protocol. It defines what the domain needs from the outside world without specifying how it's done.

# domain/ports.py
from typing import Protocol

class InventoryPort(Protocol):
    def get_stock(self, sku: str) -> int: ...
    def reserve(self, sku: str, quantity: int) -> int: ...
    def exists(self, sku: str) -> bool: ...

Now, we write the "Use Case." This is pure logic that only speaks the language of the domain. It takes the InventoryPort as an argument, allowing us to swap the real database for a simple mock during testing.

# domain/use_cases.py
def place_order(inventory: InventoryPort, request: OrderRequest) -> OrderPlaced:
    if not inventory.exists(request.sku):
        raise UnknownSKU(request.sku)
    
    available = inventory.get_stock(request.sku)
    if available < request.quantity:
        raise OutOfStock(request.sku, request.quantity, available)
        
    remaining = inventory.reserve(request.sku, request.quantity)
    return OrderPlaced(sku=request.sku, quantity=request.quantity, remaining_stock=remaining)

Implementing the Adapter

The Adapter is the glue code. The SQLAlchemyInventoryAdapter implements the InventoryPort using real database calls. The

route then becomes a thin translation layer: it converts incoming JSON to a domain OrderRequest, calls the use case, and maps domain errors back to HTTP status codes.

# adapters/sql_alchemy.py
class SQLAlchemyInventoryAdapter(InventoryPort):
    def __init__(self, conn: Connection):
        self.conn = conn

    def get_stock(self, sku: str) -> int:
        # Real SQL Alchemy logic here
        ...

Syntax Notes & Best Practices

• Structural Typing: Using typing.Protocol is preferred over abc.ABC because it allows for duck-typing. Any class with the right methods automatically satisfies the port.
• Data Classes: Use @dataclass(frozen=True) for domain models. Immutability prevents side effects within your business logic.
• Error Translation: Always catch domain errors at the edge (the API adapter) and translate them to transport-specific errors. This keeps your domain "clean."

Tips & Gotchas

Don't let the extra files intimidate you. While

adds boilerplate, it drastically reduces the "Cognitive Load" of testing. If your use case needs to be atomic, handle the transaction logic within your adapter implementation rather than polluting the domain with db.commit() calls. This keeps your architecture flexible for future changes.