Skip to content
LowLevelDesign Mastery
Start Learning Study Plan Cheat Sheets Feedback Playground

Hexagonal Architecture (Ports & Adapters)

Your business logic at the center, everything else is just a detail

By Vishnu Darshan Sanku · Updated September 25, 2026

What is Hexagonal Architecture?

Section titled “What is Hexagonal Architecture?”

The Simple Explanation (ELI5)

Section titled “The Simple Explanation (ELI5)”

Imagine your video game console:

  • The console (core): Runs your games with all the game logic
  • Ports: HDMI port, controller ports, power port
  • Adapters: HDMI cable, wireless controller, power adapter

You can swap adapters (use different TVs, controllers) without changing the console. That’s hexagonal architecture - your business logic is the console, everything else can be swapped!

Real-World Analogy: Universal Power Adapter

Section titled “Real-World Analogy: Universal Power Adapter”

Think of a universal power adapter for your laptop:

  • Core (Business Logic): Your laptop’s internal components - CPU, RAM, motherboard
  • Ports: The power input port defines what it needs (voltage, current)
  • Adapters: Different power adapters for different countries/regions
    • US adapter (110V)
    • EU adapter (220V)
    • UK adapter (240V)
    • Car adapter (12V)

Key Insight: Your laptop doesn’t care which adapter you use - as long as it provides the right voltage/current (implements the port interface), it works! You can travel anywhere and just swap the adapter.

In Software: Your business logic doesn’t care if you use PostgreSQL or MongoDB - as long as they implement the OrderRepository interface, your code works!

The Problem It Solves

Section titled “The Problem It Solves”

Traditional Layered Architecture Problem:

# BAD: Business logic depends on infrastructure
class OrderService:
    def __init__(self):
        # Tightly coupled to PostgreSQL!
        self.db = psycopg2.connect("postgresql://...")


    def create_order(self, order):
        # Business logic mixed with SQL
        cursor = self.db.cursor()
        cursor.execute("INSERT INTO orders ...")
        # What if we want to use MongoDB? Rewrite everything!

Problems:

  • Hard to test (need real database)
  • Hard to swap databases (tightly coupled)
  • Business logic mixed with infrastructure code
  • Can’t run without database running

Hexagonal Architecture Solution:

# GOOD: Business logic depends on interfaces
class OrderService:
    def __init__(self, order_repo: OrderRepository):  # Interface!
        self._repo = order_repo  # Could be PostgreSQL, MongoDB, or Mock!


    def create_order(self, order):
        # Pure business logic - no infrastructure!
        if order.total < 0:
            raise ValueError("Invalid order")
        return self._repo.save(order)  # Just calls interface

Benefits:

  • Easy to test (use mock repository)
  • Easy to swap databases (just change adapter)
  • Business logic isolated from infrastructure
  • Can run without database (use in-memory adapter)

Why “Hexagonal”?

Section titled “Why “Hexagonal”?”

The shape is arbitrary - Alistair Cockburn chose a hexagon because:

  1. It has enough sides to draw multiple ports
  2. It’s visually distinct from circles/rectangles
  3. It doesn’t imply any specific number of ports (you can have 3, 6, 10 ports)

It could be called:

  • Octagonal Architecture (8 sides)
  • Polygonal Architecture (many sides)
  • Ports & Adapters Architecture (more accurate name!)

The important part is the concept, not the shape!

Core Principles

Section titled “Core Principles”

1. Business Logic at the Center

  • All business rules, domain entities, use cases live in the core
  • Zero dependencies on external frameworks or libraries
  • Pure, testable code

2. Ports Define Contracts

  • Primary Ports: What the application offers (use cases, APIs)
  • Secondary Ports: What the application needs (databases, external services)
  • Ports are interfaces - they define what, not how

3. Adapters Implement Ports

  • Primary Adapters: Implement how external systems call your app (REST, GraphQL, CLI)
  • Secondary Adapters: Implement how your app calls external systems (PostgreSQL, Redis, Email)
  • Adapters are pluggable - swap them without changing core

4. Dependencies Point Inward

  • Core has zero dependencies on adapters
  • Adapters depend on core (ports)
  • This is Dependency Inversion Principle in action!

Real-World Example: E-Commerce Application

Section titled “Real-World Example: E-Commerce Application”

Without Hexagonal Architecture:

Web Controller → OrderService → PostgreSQL
                    ↓
                 (tightly coupled!)

Problems:

  • Can’t test without PostgreSQL running
  • Can’t switch to MongoDB without rewriting OrderService
  • Business logic mixed with SQL queries

With Hexagonal Architecture:

REST Adapter → OrderService → OrderRepository (port)
GraphQL Adapter →    ↓         ↓
CLI Adapter →    (core)    PostgreSQL Adapter
                         MongoDB Adapter
                         In-Memory Adapter (tests)

Benefits:

  • Test with in-memory adapter (no database needed)
  • Switch databases by swapping adapter
  • Add GraphQL API without changing business logic
  • Business logic completely isolated

When to Use Hexagonal Architecture

Section titled “When to Use Hexagonal Architecture”

Use When:

  • You need to test business logic in isolation
  • You want to swap technologies (databases, frameworks) easily
  • You have multiple interfaces (REST, GraphQL, CLI)
  • You want technology independence
  • You’re building long-lived applications that will evolve

Don’t Use When:

  • Simple CRUD applications with one interface
  • Prototypes/MVPs where speed matters more than flexibility
  • Applications that will never change technologies
  • Team doesn’t understand the pattern (adds complexity)

Real-World Usage:

  • Netflix: Uses hexagonal architecture for microservices
  • Amazon: Many services use ports/adapters pattern
  • Uber: Core business logic isolated from infrastructure
  • Banking Systems: Need to swap databases, add new interfaces

The Hexagonal Structure

Section titled “The Hexagonal Structure”
Hexagonal architecture with REST, GraphQL and CLI adapters driving core ports, and Postgres, Redis, Kafka and email adapters driven by it

Ports and Adapters Explained

Section titled “Ports and Adapters Explained”

Ports (Interfaces)

Section titled “Ports (Interfaces)”

A port is an interface that defines HOW to interact with the application core.

Two Types:

  1. Primary Ports (Driving Ports)

    • Purpose: What the application offers to the outside world
    • Direction: External → Core (adapters CALL these)
    • Examples: Use case interfaces, command handlers

  2. Secondary Ports (Driven Ports)

    • Purpose: What the application needs from the outside world
    • Direction: Core → External (adapters IMPLEMENT these)
    • Examples: Repository interfaces, notification interfaces
Ports and adapters flow from external world through a primary port into the application core, then a secondary port to infrastructure

Adapters (Implementations)

Section titled “Adapters (Implementations)”

An adapter is a concrete implementation of a port for a specific technology.

Two Types:

  1. Primary Adapters (Driving Adapters)

    • Convert external requests → port calls
    • Examples: REST controller, GraphQL resolver, CLI command

  2. Secondary Adapters (Driven Adapters)

    • Implement ports using specific technology
    • Examples: PostgreSQL repository, SendGrid email sender

Hexagonal Architecture in Code

Section titled “Hexagonal Architecture in Code”

Example: Order Management System

Section titled “Example: Order Management System”

1. Application Core (The Hexagon)

Section titled “1. Application Core (The Hexagon)”
💡 Tip: Click dropdown to switch between languages
core/domain/order.py
from dataclasses import dataclass
from decimal import Decimal
from enum import Enum
from typing import List


class OrderStatus(Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    CANCELLED = "cancelled"


@dataclass
class OrderItem:
    product_id: str
    quantity: int
    price: Decimal


class Order:
    """Domain entity - pure business logic"""


    def __init__(self, id: str, customer_id: str):
        self.id = id
        self.customer_id = customer_id
        self.items: List[OrderItem] = []
        self.status = OrderStatus.PENDING


    def add_item(self, product_id: str, quantity: int, price: Decimal):
        """Business rule: Cannot add items to confirmed order"""
        if self.status == OrderStatus.CONFIRMED:
            raise ValueError("Cannot modify confirmed order")


        self.items.append(OrderItem(product_id, quantity, price))


    def calculate_total(self) -> Decimal:
        return sum(item.price * item.quantity for item in self.items)


    def confirm(self):
        """Business rule: Must have items to confirm"""
        if not self.items:
            raise ValueError("Cannot confirm empty order")
        if self.status != OrderStatus.PENDING:
            raise ValueError("Order already confirmed")


        self.status = OrderStatus.CONFIRMED

2. Primary Port (What App Offers)

Section titled “2. Primary Port (What App Offers)”
💡 Tip: Click dropdown to switch between languages
core/ports/primary/order_use_cases.py
from abc import ABC, abstractmethod
from typing import List


class OrderUseCases(ABC):
    """Primary port - defines what the application offers"""


    @abstractmethod
    async def place_order(
        self,
        customer_id: str,
        items: List[dict]
    ) -> Order:
        """Place a new order"""
        pass


    @abstractmethod
    async def confirm_order(self, order_id: str) -> Order:
        """Confirm an order"""
        pass


    @abstractmethod
    async def cancel_order(self, order_id: str, reason: str) -> Order:
        """Cancel an order"""
        pass


    @abstractmethod
    async def get_order(self, order_id: str) -> Order:
        """Retrieve order by ID"""
        pass

3. Secondary Ports (What App Needs)

Section titled “3. Secondary Ports (What App Needs)”
💡 Tip: Click dropdown to switch between languages
core/ports/secondary/order_repository.py
from abc import ABC, abstractmethod
from typing import Optional


class OrderRepository(ABC):
    """Secondary port - defines what the application needs for persistence"""


    @abstractmethod
    async def save(self, order: Order) -> None:
        """Persist an order"""
        pass


    @abstractmethod
    async def find_by_id(self, order_id: str) -> Optional[Order]:
        """Find order by ID"""
        pass


class NotificationService(ABC):
    """Secondary port - defines what the application needs for notifications"""


    @abstractmethod
    async def notify_order_confirmed(self, order: Order) -> None:
        """Send confirmation notification"""
        pass


    @abstractmethod
    async def notify_order_cancelled(self, order: Order, reason: str) -> None:
        """Send cancellation notification"""
        pass


class PaymentService(ABC):
    """Secondary port - defines what the application needs for payments"""


    @abstractmethod
    async def process_payment(self, order_id: str, amount: Decimal) -> bool:
        """Process payment for order"""
        pass

4. Use Case Implementation (Application Core)

Section titled “4. Use Case Implementation (Application Core)”
💡 Tip: Click dropdown to switch between languages
core/application/order_service.py
class OrderService(OrderUseCases):
    """
    Implements primary port (what app offers)
    Uses secondary ports (what app needs)
    """


    def __init__(
        self,
        order_repo: OrderRepository,
        payment_service: PaymentService,
        notification_service: NotificationService
    ):
        # Depend on INTERFACES (ports), not implementations!
        self._order_repo = order_repo
        self._payment = payment_service
        self._notifications = notification_service


    async def place_order(
        self,
        customer_id: str,
        items: List[dict]
    ) -> Order:
        """Use case implementation"""


        # Create order (domain logic)
        order = Order(id=generate_id(), customer_id=customer_id)


        for item in items:
            order.add_item(
                product_id=item['product_id'],
                quantity=item['quantity'],
                price=Decimal(item['price'])
            )


        # Save order (use secondary port)
        await self._order_repo.save(order)


        return order


    async def confirm_order(self, order_id: str) -> Order:
        """Use case implementation"""


        # Load order (use secondary port)
        order = await self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"Order {order_id} not found")


        # Confirm order (domain logic)
        order.confirm()


        # Process payment (use secondary port)
        payment_success = await self._payment.process_payment(
            order.id,
            order.calculate_total()
        )


        if not payment_success:
            raise ValueError("Payment failed")


        # Save confirmed order (use secondary port)
        await self._order_repo.save(order)


        # Send notification (use secondary port)
        await self._notifications.notify_order_confirmed(order)


        return order


    async def cancel_order(self, order_id: str, reason: str) -> Order:
        """Use case implementation"""


        order = await self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"Order {order_id} not found")


        order.cancel()
        await self._order_repo.save(order)
        await self._notifications.notify_order_cancelled(order, reason)


        return order


    async def get_order(self, order_id: str) -> Order:
        """Use case implementation"""
        order = await self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"Order {order_id} not found")
        return order

5. Primary Adapter (REST API)

Section titled “5. Primary Adapter (REST API)”
💡 Tip: Click dropdown to switch between languages
adapters/primary/rest_api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel


app = FastAPI()


class CreateOrderRequest(BaseModel):
    customer_id: str
    items: list[dict]


class OrderResponse(BaseModel):
    order_id: str
    status: str
    total: float


@app.post("/orders", response_model=OrderResponse)
async def create_order(request: CreateOrderRequest):
    """Primary adapter - converts HTTP → port call"""
    try:
        # Call primary port
        order = await order_use_cases.place_order(
            customer_id=request.customer_id,
            items=request.items
        )


        # Convert domain model → DTO
        return OrderResponse(
            order_id=order.id,
            status=order.status.value,
            total=float(order.calculate_total())
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))


@app.post("/orders/{order_id}/confirm")
async def confirm_order(order_id: str):
    """Primary adapter - converts HTTP → port call"""
    try:
        order = await order_use_cases.confirm_order(order_id)
        return OrderResponse(
            order_id=order.id,
            status=order.status.value,
            total=float(order.calculate_total())
        )
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))

6. Secondary Adapter (PostgreSQL Repository)

Section titled “6. Secondary Adapter (PostgreSQL Repository)”
💡 Tip: Click dropdown to switch between languages
adapters/secondary/postgres_order_repository.py
from sqlalchemy.orm import Session
from typing import Optional


class PostgresOrderRepository(OrderRepository):
    """Secondary adapter - implements port using PostgreSQL"""


    def __init__(self, db_session: Session):
        self._session = db_session


    async def save(self, order: Order) -> None:
        """Convert domain → database model"""
        db_order = OrderModel(
            id=order.id,
            customer_id=order.customer_id,
            status=order.status.value
        )


        # Save order items
        for item in order.items:
            db_item = OrderItemModel(
                order_id=order.id,
                product_id=item.product_id,
                quantity=item.quantity,
                price=item.price
            )
            db_order.items.append(db_item)


        self._session.add(db_order)
        await self._session.commit()


    async def find_by_id(self, order_id: str) -> Optional[Order]:
        """Convert database model → domain"""
        db_order = await self._session.query(OrderModel)\
            .filter_by(id=order_id)\
            .first()


        if not db_order:
            return None


        # Reconstruct domain entity
        order = Order(
            id=db_order.id,
            customer_id=db_order.customer_id
        )
        order.status = OrderStatus(db_order.status)


        for db_item in db_order.items:
            order.items.append(OrderItem(
                product_id=db_item.product_id,
                quantity=db_item.quantity,
                price=db_item.price
            ))


        return order

7. Secondary Adapter (Email Notification)

Section titled “7. Secondary Adapter (Email Notification)”
💡 Tip: Click dropdown to switch between languages
adapters/secondary/email_notification_service.py
import smtplib
from email.message import EmailMessage


class EmailNotificationService(NotificationService):
    """Secondary adapter - implements port using SMTP"""


    def __init__(self, smtp_host: str, smtp_port: int):
        self._host = smtp_host
        self._port = smtp_port


    async def notify_order_confirmed(self, order: Order) -> None:
        """Send confirmation email"""
        message = EmailMessage()
        message['Subject'] = f'Order {order.id} Confirmed'
        message['To'] = await self._get_customer_email(order.customer_id)
        message.set_content(
            f'Your order {order.id} has been confirmed.\n'
            f'Total: ${order.calculate_total()}'
        )


        with smtplib.SMTP(self._host, self._port) as smtp:
            smtp.send_message(message)


    async def notify_order_cancelled(
        self,
        order: Order,
        reason: str
    ) -> None:
        """Send cancellation email"""
        message = EmailMessage()
        message['Subject'] = f'Order {order.id} Cancelled'
        message['To'] = await self._get_customer_email(order.customer_id)
        message.set_content(
            f'Your order {order.id} has been cancelled.\n'
            f'Reason: {reason}'
        )


        with smtplib.SMTP(self._host, self._port) as smtp:
            smtp.send_message(message)

Advantages of Hexagonal Architecture

Section titled “Advantages of Hexagonal Architecture”

Technology Independence

Swap databases, change UI framework, switch messaging systems - all without touching business logic!

Testability

Test business logic in isolation. Mock all adapters. No need for database or HTTP server in tests.

Flexibility

Add new adapters easily. Support REST and GraphQL simultaneously. Use multiple databases.

Clear Boundaries

Business logic is completely isolated. No accidental dependencies on infrastructure.

Hexagonal vs Layered Architecture

Section titled “Hexagonal vs Layered Architecture”
AspectLayered ArchitectureHexagonal Architecture
StructureHorizontal layersCenter + adapters
DependenciesTop-down through layersAll point inward to core
FlexibilityLess flexibleMore flexible
InfrastructureBottom layerAdapters on outside
TestingTest through layersTest core directly
Swap ComponentsHarderEasier
ComplexitySimplerMore complex
Best ForTraditional appsEvent-driven, multiple interfaces

Key Takeaways

Section titled “Key Takeaways”

Business Logic First

Core is pure business logic with zero infrastructure dependencies. Everything else is just an implementation detail!

Ports Define Contracts

Interfaces (ports) define HOW to interact with the core. Adapters provide specific implementations.

Easily Swappable

Change databases, UI frameworks, or messaging systems without touching business logic. True technology independence!

Testing Paradise

Test business logic without database, HTTP, or any external dependencies. Fast, reliable, isolated tests.

Section titled “Related Topics”

Further Reading

Section titled “Further Reading”
  • “Hexagonal Architecture” by Alistair Cockburn (original article)
  • “Get Your Hands Dirty on Clean Architecture” by Tom Hombergs
  • “Clean Architecture” by Robert C. Martin
  • “Implementing Domain-Driven Design” by Vaughn Vernon