Ports and Adapters: The Pattern Your Architecture Has Been Missing

In the last post you counted your OrderService's external dependencies. If you got anything higher than zero, this post is for you.

We named the problem. Now let's name the solution.

---

It Has a Name

The pattern is called ports and adapters also known as hexagonal architecture, coined by Alistair Cockburn in 2005. The name has stuck around because the idea is genuinely useful: your application core should communicate with the outside world through well-defined boundaries, not direct dependencies.

That's the theory. Here's what it actually means in code.

---

What a Port Is

A port is an interface owned by your domain. It says: "I need something that can do X. I don't care how."

from typing import Protocol, Optional
from app.domain.models import Order, Customer

class CustomerRepository(Protocol):
    def find_by_id(self, customer_id: str) -> Optional[Customer]:
        ...

class OrderRepository(Protocol):
    def save(self, order: Order) -> Order:
        ...

Notice what these interfaces don't mention: SQLAlchemy, PostgreSQL, sessions, ORM models. They speak the language of your domain — Customer, Order not the language of your database.

This is the port. It's the socket in your domain that infrastructure plugs into.

---

What an Adapter Is

An adapter is the infrastructure implementation of a port. It's the plug.

from sqlalchemy.orm import Session
from app.domain.ports import CustomerRepository, OrderRepository
from app.domain.models import Customer, Order
from app import orm

class SQLAlchemyCustomerRepository:
    def __init__(self, session: Session):
        self.session = session

    def find_by_id(self, customer_id: str) -> Optional[Customer]:
        row = self.session.query(orm.CustomerRow).filter_by(id=customer_id).first()
        if not row:
            return None
        return Customer(
            id=row.id,
            loyalty_points=row.loyalty_points
        )

class SQLAlchemyOrderRepository:
    def __init__(self, session: Session):
        self.session = session

    def save(self, order: Order) -> Order:
        row = orm.OrderRow(
            customer_id=order.customer_id,
            total=order.total,
            status=order.status
        )
        self.session.add(row)
        self.session.commit()
        self.session.refresh(row)
        order.id = row.id
        return order

SQLAlchemy lives here. The ORM mapping lives here. The session.commit() lives here. Your domain doesn't see any of it.

---

The Before/After

Here's the OrderService from Post 1 the one that knew too much:

class OrderService:
    def place_order(self, customer_id: int, items: list, db: Session):
        customer = db.query(Customer).filter(Customer.id == customer_id).first()
        if not customer:
            raise HTTPException(status_code=404, detail="Customer not found")

        total = sum(item["price"] * item["qty"] for item in items)
        if customer.loyalty_points > 500:
            total *= 0.9

        order = Order(customer_id=customer_id, total=total, status="pending", items=items)
        db.add(order)
        db.commit()
        db.refresh(order)
        return {"order_id": order.id, "total": total, "status": "pending"}

Here's the same service after the refactor:

from app.domain.ports import CustomerRepository, OrderRepository
from app.domain.models import Order, OrderItem

class OrderService:
    def __init__(
        self,
        customer_repo: CustomerRepository,
        order_repo: OrderRepository,
    ):
        self.customer_repo = customer_repo
        self.order_repo = order_repo

    def place_order(self, customer_id: str, items: list[OrderItem]) -> Order:
        customer = self.customer_repo.find_by_id(customer_id)
        if not customer:
            raise ValueError(f"Customer {customer_id} not found")

        order = Order(customer_id=customer_id, items=items, loyalty_points=customer.loyalty_points)
        order.total = order.calculate_total()
        order.status = "pending"

        return self.order_repo.save(order)

What changed:

• No db: Session parameter. The service doesn't know a database exists.
• No HTTPException. That's the API layer's job, not the domain's.
• No ORM models. Order and Customer are plain domain objects.
• The dependencies come in through __init__ the service declares what it needs without knowing how it's provided.

---

Now Try to Test It

class InMemoryCustomerRepository:
    def __init__(self, customers: dict):
        self.customers = customers

    def find_by_id(self, customer_id: str) -> Optional[Customer]:
        return self.customers.get(customer_id)

class InMemoryOrderRepository:
    def __init__(self):
        self.saved = []

    def save(self, order: Order) -> Order:
        order.id = f"order-{len(self.saved) + 1}"
        self.saved.append(order)
        return order


def test_loyalty_discount_applied():
    customer_repo = InMemoryCustomerRepository({
        "c-1": Customer(id="c-1", loyalty_points=600)
    })
    order_repo = InMemoryOrderRepository()
    service = OrderService(customer_repo, order_repo)

    result = service.place_order("c-1", [OrderItem("p-1", 100.0, 2)])

    assert result.total == 180.0
    assert result.status == "pending"

No database. No HTTP framework. No environment variables. The test runs in milliseconds and covers exactly what it claims to cover.

This is the payoff.

---

The Direction of Dependency

Here's the rule that makes this work: dependencies point inward.

FastAPI route → OrderService → CustomerRepository (port)
                                      ↑
                     SQLAlchemyCustomerRepository (adapter) ← FastAPI route wires this in

The domain defines the port. The adapter implements it. The wiring happens at the edges in your FastAPI startup, your dependency injection container, wherever your app is assembled. The domain never reaches out to the infrastructure. The infrastructure reaches in to fulfill the domain's contracts.

This is the inversion of control at the heart of the pattern.

---

What You Get

Swappable infrastructure. Want to test against an in-memory store? Swap the adapter. Migrating from PostgreSQL to another database? Write a new adapter. The domain doesn't notice.

Testable business logic. No fixtures. No Docker containers in unit tests. The domain logic tests itself.

Explicit boundaries. Every time someone wants to add a database call to the service layer, they have to go through the port. The port is a checkpoint. It forces the question: is this a domain concern, or infrastructure?

---

One Thing to Watch Out For

Ports should be defined by what the domain needs, not by what the database offers. A port that looks like this is a red flag:

class OrderRepository(Protocol):
    def execute_raw_sql(self, query: str) -> list:
        ...

That's an infrastructure interface wearing a domain name. The domain shouldn't know about SQL. Keep ports at the level of domain operations: find_by_id, save, find_pending_orders. If you can read the method name and understand the business intent without knowing anything about databases, you're in the right place.

---

What's Next

You now have the pattern. Ports declare the need. Adapters fulfill it. The domain stays clean.

But ports and adapters only work if you have a domain worth protecting. If your domain objects are just bags of getters and setters no behavior, no rules, no real logic the boundaries don't buy you much.

The next post is about fixing that. We're going to build a domain model that actually models the domain: real behavior on real objects, not data containers with a fancy name.

---

This is Post 2 of 5 in the Hexagonal Architecture series.