Hexagonal Architecture (Ports and Adapters)

Study this system. It works today. Then your team is asked to (a) replace PostgreSQL with DynamoDB, (b) add a CLI interface alongside the REST API, and (c) swap SendGrid for AWS SES. How many files change?

# app.py - everything in one place
from fastapi import FastAPI
from sqlalchemy import create_engine, text
import sendgrid

app = FastAPI()
engine = create_engine("postgresql://localhost/mydb")
sg = sendgrid.SendGridAPIClient("SG.xxx")

@app.post("/notifications")
def send_notification(user_id: int, message: str):
    with engine.connect() as conn:
        row = conn.execute(
            text("SELECT email FROM users WHERE id = :id"), {"id": user_id}
        ).fetchone()
    if not row:
        return {"error": "User not found"}
    sg.send(
        sendgrid.Mail(
            from_email="[email protected]",
            to_emails=row.email,
            subject="Notification",
            plain_text_content=message,
        )
    )
    return {"status": "sent"}

Answer: every single line. The database query, the email sending, and the HTTP handling are fused together. Hexagonal Architecture eliminates this coupling by introducing ports (what the application needs) and adapters (how those needs are fulfilled).

What You Will Learn

• The hexagonal model: application core surrounded by ports and adapters
• Primary (driving) ports vs secondary (driven) ports and why the distinction matters
• Implementing ports as Python Protocol classes
• Building adapters that can be swapped without touching business logic
• Wiring a complete FastAPI + SQLAlchemy application using hexagonal architecture
• Testing strategies that exploit adapter swappability
• How hexagonal architecture relates to and differs from Clean Architecture

Prerequisites

• Understanding of Clean Architecture layers and the dependency rule (previous lesson)
• Familiarity with typing.Protocol for structural subtyping
• Experience with FastAPI dependency injection and SQLAlchemy ORM
• Comfort with dataclasses and abstract interfaces

Part 1 - The Hexagonal Model

Alistair Cockburn introduced hexagonal architecture in 2005 to solve a specific problem: applications were being written so that business logic could not be tested or run without their UI, database, and external services.

The solution is to structure the application as a hexagon (the shape is metaphorical - it represents "many sides" where adapters can plug in):

Primary vs Secondary Ports

Aspect	Primary (Driving) Ports	Secondary (Driven) Ports
Direction	Outside drives the application	Application drives the outside
Who calls whom	Adapter calls the port	Application calls the port
Examples	HTTP endpoints, CLI commands, event handlers	Database, email, payment gateway, cache
Implementation	The application implements these	The adapters implement these
Testing	You write tests that call primary ports	You provide fake adapters for secondary ports

This asymmetry is crucial. Primary ports define what the application offers. Secondary ports define what the application needs.

Part 2 - Defining Ports with Protocol

Ports are interfaces. In Python, Protocol provides structural subtyping - any class that has the right methods satisfies the port, without explicit inheritance.

Secondary Ports (What the Application Needs)

# ports/outbound.py
from typing import Protocol, Optional, runtime_checkable
from uuid import UUID
from datetime import datetime
from dataclasses import dataclass


@dataclass
class Article:
    """Domain entity - no framework dependencies."""
    id: UUID
    title: str
    content: str
    author_id: UUID
    published_at: Optional[datetime] = None
    is_published: bool = False

    def publish(self) -> None:
        if not self.title or not self.content:
            raise ValueError("Cannot publish article without title and content")
        self.is_published = True
        self.published_at = datetime.utcnow()

    def unpublish(self) -> None:
        self.is_published = False
        self.published_at = None


@runtime_checkable
class ArticleRepository(Protocol):
    """Secondary port: the application needs to store and retrieve articles."""

    def save(self, article: Article) -> None: ...
    def get_by_id(self, article_id: UUID) -> Optional[Article]: ...
    def list_published(self, limit: int = 20, offset: int = 0) -> list[Article]: ...
    def delete(self, article_id: UUID) -> None: ...


@runtime_checkable
class SearchIndex(Protocol):
    """Secondary port: the application needs to index articles for search."""

    def index_article(self, article: Article) -> None: ...
    def remove_article(self, article_id: UUID) -> None: ...
    def search(self, query: str, limit: int = 10) -> list[UUID]: ...


@runtime_checkable
class NotificationService(Protocol):
    """Secondary port: the application needs to send notifications."""

    def notify_author(self, author_id: UUID, message: str) -> None: ...
    def notify_subscribers(self, article: Article) -> None: ...


@runtime_checkable
class EventBus(Protocol):
    """Secondary port: the application needs to publish domain events."""

    def publish(self, event_type: str, payload: dict) -> None: ...

Primary Ports (What the Application Offers)

# ports/inbound.py
from typing import Protocol
from uuid import UUID
from dataclasses import dataclass


@dataclass
class CreateArticleCommand:
    title: str
    content: str
    author_id: UUID


@dataclass
class PublishArticleCommand:
    article_id: UUID
    actor_id: UUID


@dataclass
class ArticleResult:
    id: UUID
    title: str
    is_published: bool


class ArticleManagement(Protocol):
    """Primary port: what the application offers to the outside world."""

    def create_article(self, command: CreateArticleCommand) -> ArticleResult: ...
    def publish_article(self, command: PublishArticleCommand) -> ArticleResult: ...
    def get_article(self, article_id: UUID) -> ArticleResult: ...
    def list_articles(self, limit: int, offset: int) -> list[ArticleResult]: ...

:::note Why @runtime_checkable? Adding @runtime_checkable lets you verify at runtime that an adapter satisfies a port: isinstance(my_adapter, ArticleRepository). This is useful for assertions during wiring but is not required for type checking - mypy and pyright check Protocol conformance statically. :::

Part 3 - Building the Application Core

The application core implements the primary ports and depends on the secondary ports through constructor injection.

# core/article_service.py
from uuid import UUID, uuid4
from ports.outbound import (
    Article,
    ArticleRepository,
    EventBus,
    NotificationService,
    SearchIndex,
)
from ports.inbound import (
    ArticleManagement,
    ArticleResult,
    CreateArticleCommand,
    PublishArticleCommand,
)


class ArticleService:
    """
    Application core - implements the primary port (ArticleManagement)
    and depends on secondary ports via constructor injection.
    """

    def __init__(
        self,
        repo: ArticleRepository,
        search: SearchIndex,
        notifications: NotificationService,
        events: EventBus,
    ) -> None:
        self._repo = repo
        self._search = search
        self._notifications = notifications
        self._events = events

    def create_article(self, command: CreateArticleCommand) -> ArticleResult:
        article = Article(
            id=uuid4(),
            title=command.title,
            content=command.content,
            author_id=command.author_id,
        )
        self._repo.save(article)
        self._events.publish("article.created", {"id": str(article.id)})

        return ArticleResult(
            id=article.id,
            title=article.title,
            is_published=article.is_published,
        )

    def publish_article(self, command: PublishArticleCommand) -> ArticleResult:
        article = self._repo.get_by_id(command.article_id)
        if article is None:
            raise ValueError(f"Article {command.article_id} not found")

        if article.author_id != command.actor_id:
            raise PermissionError("Only the author can publish an article")

        article.publish()  # domain logic - validates title/content

        self._repo.save(article)
        self._search.index_article(article)
        self._notifications.notify_subscribers(article)
        self._events.publish("article.published", {"id": str(article.id)})

        return ArticleResult(
            id=article.id,
            title=article.title,
            is_published=article.is_published,
        )

    def get_article(self, article_id: UUID) -> ArticleResult:
        article = self._repo.get_by_id(article_id)
        if article is None:
            raise ValueError(f"Article {article_id} not found")
        return ArticleResult(
            id=article.id,
            title=article.title,
            is_published=article.is_published,
        )

    def list_articles(self, limit: int = 20, offset: int = 0) -> list[ArticleResult]:
        articles = self._repo.list_published(limit, offset)
        return [
            ArticleResult(id=a.id, title=a.title, is_published=a.is_published)
            for a in articles
        ]

Notice what ArticleService does not import: no SQLAlchemy, no FastAPI, no Elasticsearch client, no email library. It speaks only in terms of domain entities and port interfaces.

Part 4 - Building Adapters

Adapters are the concrete implementations that plug into ports. Each adapter is independent and replaceable.

Database Adapter (SQLAlchemy)

# adapters/outbound/sqlalchemy_article_repo.py
from typing import Optional
from uuid import UUID

from sqlalchemy import Column, String, DateTime, Boolean, Text
from sqlalchemy.orm import Session, declarative_base

from ports.outbound import Article

Base = declarative_base()


class ArticleModel(Base):
    __tablename__ = "articles"

    id = Column(String, primary_key=True)
    title = Column(String(255), nullable=False)
    content = Column(Text, nullable=False)
    author_id = Column(String, nullable=False, index=True)
    published_at = Column(DateTime, nullable=True)
    is_published = Column(Boolean, default=False)


class SqlAlchemyArticleRepository:
    """Adapter: implements the ArticleRepository port using SQLAlchemy."""

    def __init__(self, session: Session) -> None:
        self._session = session

    def save(self, article: Article) -> None:
        model = ArticleModel(
            id=str(article.id),
            title=article.title,
            content=article.content,
            author_id=str(article.author_id),
            published_at=article.published_at,
            is_published=article.is_published,
        )
        self._session.merge(model)
        self._session.flush()

    def get_by_id(self, article_id: UUID) -> Optional[Article]:
        row = self._session.get(ArticleModel, str(article_id))
        if row is None:
            return None
        return self._to_domain(row)

    def list_published(self, limit: int = 20, offset: int = 0) -> list[Article]:
        rows = (
            self._session.query(ArticleModel)
            .filter(ArticleModel.is_published.is_(True))
            .order_by(ArticleModel.published_at.desc())
            .offset(offset)
            .limit(limit)
            .all()
        )
        return [self._to_domain(r) for r in rows]

    def delete(self, article_id: UUID) -> None:
        row = self._session.get(ArticleModel, str(article_id))
        if row:
            self._session.delete(row)
            self._session.flush()

    @staticmethod
    def _to_domain(row: ArticleModel) -> Article:
        return Article(
            id=UUID(row.id),
            title=row.title,
            content=row.content,
            author_id=UUID(row.author_id),
            published_at=row.published_at,
            is_published=row.is_published,
        )

Search Adapter (Elasticsearch)

# adapters/outbound/elasticsearch_search.py
from uuid import UUID
from elasticsearch import Elasticsearch
from ports.outbound import Article


class ElasticsearchSearchIndex:
    """Adapter: implements the SearchIndex port using Elasticsearch."""

    def __init__(self, client: Elasticsearch, index_name: str = "articles") -> None:
        self._client = client
        self._index = index_name

    def index_article(self, article: Article) -> None:
        self._client.index(
            index=self._index,
            id=str(article.id),
            document={
                "title": article.title,
                "content": article.content,
                "author_id": str(article.author_id),
                "published_at": (
                    article.published_at.isoformat() if article.published_at else None
                ),
            },
        )

    def remove_article(self, article_id: UUID) -> None:
        self._client.delete(index=self._index, id=str(article_id), ignore=[404])

    def search(self, query: str, limit: int = 10) -> list[UUID]:
        result = self._client.search(
            index=self._index,
            query={"multi_match": {"query": query, "fields": ["title", "content"]}},
            size=limit,
        )
        return [UUID(hit["_id"]) for hit in result["hits"]["hits"]]

Notification Adapter (Email via SMTP)

# adapters/outbound/email_notifications.py
import smtplib
from email.mime.text import MIMEText
from uuid import UUID
from ports.outbound import Article


class EmailNotificationService:
    """Adapter: implements the NotificationService port using SMTP."""

    def __init__(
        self,
        smtp_host: str,
        smtp_port: int,
        from_address: str,
        subscriber_lookup: callable,
    ) -> None:
        self._host = smtp_host
        self._port = smtp_port
        self._from = from_address
        self._subscriber_lookup = subscriber_lookup

    def notify_author(self, author_id: UUID, message: str) -> None:
        # In production, look up author email from user service
        pass

    def notify_subscribers(self, article: Article) -> None:
        emails = self._subscriber_lookup(article.author_id)
        for email in emails:
            msg = MIMEText(f"New article published: {article.title}")
            msg["Subject"] = f"New: {article.title}"
            msg["From"] = self._from
            msg["To"] = email
            with smtplib.SMTP(self._host, self._port) as server:
                server.send_message(msg)

Event Bus Adapter (Redis Pub/Sub)

# adapters/outbound/redis_events.py
import json
import redis


class RedisEventBus:
    """Adapter: implements the EventBus port using Redis Pub/Sub."""

    def __init__(self, redis_client: redis.Redis, channel_prefix: str = "app") -> None:
        self._redis = redis_client
        self._prefix = channel_prefix

    def publish(self, event_type: str, payload: dict) -> None:
        channel = f"{self._prefix}.{event_type}"
        self._redis.publish(channel, json.dumps(payload))

Part 5 - Driving Adapters (Primary Side)

Driving adapters call the application's primary ports. They translate external input formats into commands the application understands.

FastAPI Adapter

# adapters/inbound/fastapi_adapter.py
from uuid import UUID

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel

from core.article_service import ArticleService
from ports.inbound import CreateArticleCommand, PublishArticleCommand


router = APIRouter(prefix="/articles", tags=["articles"])


class CreateArticleRequest(BaseModel):
    title: str
    content: str
    author_id: UUID


class ArticleResponse(BaseModel):
    id: UUID
    title: str
    is_published: bool


def get_article_service() -> ArticleService:
    """Wired at application startup - see composition root."""
    ...


@router.post("/", response_model=ArticleResponse, status_code=201)
def create_article(
    body: CreateArticleRequest,
    service: ArticleService = Depends(get_article_service),
):
    try:
        result = service.create_article(
            CreateArticleCommand(
                title=body.title,
                content=body.content,
                author_id=body.author_id,
            )
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    return ArticleResponse(
        id=result.id, title=result.title, is_published=result.is_published
    )


@router.post("/{article_id}/publish", response_model=ArticleResponse)
def publish_article(
    article_id: UUID,
    actor_id: UUID,
    service: ArticleService = Depends(get_article_service),
):
    try:
        result = service.publish_article(
            PublishArticleCommand(article_id=article_id, actor_id=actor_id)
        )
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))
    except PermissionError as e:
        raise HTTPException(status_code=403, detail=str(e))
    return ArticleResponse(
        id=result.id, title=result.title, is_published=result.is_published
    )


@router.get("/", response_model=list[ArticleResponse])
def list_articles(
    limit: int = 20,
    offset: int = 0,
    service: ArticleService = Depends(get_article_service),
):
    results = service.list_articles(limit, offset)
    return [
        ArticleResponse(id=r.id, title=r.title, is_published=r.is_published)
        for r in results
    ]

CLI Adapter

# adapters/inbound/cli_adapter.py
import click
from uuid import UUID

from core.article_service import ArticleService
from ports.inbound import CreateArticleCommand, PublishArticleCommand


def create_cli(service: ArticleService) -> click.Group:
    @click.group()
    def cli():
        """Article management CLI."""
        pass

    @cli.command()
    @click.option("--title", required=True)
    @click.option("--content", required=True)
    @click.option("--author-id", required=True, type=click.UUID)
    def create(title: str, content: str, author_id: UUID):
        result = service.create_article(
            CreateArticleCommand(title=title, content=content, author_id=author_id)
        )
        click.echo(f"Created article: {result.id}")

    @cli.command()
    @click.argument("article_id", type=click.UUID)
    @click.option("--actor-id", required=True, type=click.UUID)
    def publish(article_id: UUID, actor_id: UUID):
        try:
            result = service.publish_article(
                PublishArticleCommand(article_id=article_id, actor_id=actor_id)
            )
            click.echo(f"Published: {result.title}")
        except (ValueError, PermissionError) as e:
            click.echo(f"Error: {e}", err=True)

    @cli.command("list")
    @click.option("--limit", default=20)
    def list_articles(limit: int):
        results = service.list_articles(limit, 0)
        for r in results:
            status = "published" if r.is_published else "draft"
            click.echo(f"[{status}] {r.title} ({r.id})")

    return cli

The same ArticleService powers both the REST API and the CLI. No business logic is duplicated.

Part 6 - The Composition Root

The composition root is where you wire everything together. It is the only place that knows about all concrete implementations.

# main.py - the composition root
from elasticsearch import Elasticsearch
from redis import Redis
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI

from adapters.inbound.fastapi_adapter import router, get_article_service
from adapters.outbound.elasticsearch_search import ElasticsearchSearchIndex
from adapters.outbound.email_notifications import EmailNotificationService
from adapters.outbound.redis_events import RedisEventBus
from adapters.outbound.sqlalchemy_article_repo import SqlAlchemyArticleRepository
from core.article_service import ArticleService


def create_app() -> FastAPI:
    # --- Infrastructure ---
    engine = create_engine("postgresql://user:pass@localhost/articles")
    SessionLocal = sessionmaker(bind=engine)
    es_client = Elasticsearch("http://localhost:9200")
    redis_client = Redis(host="localhost", port=6379)

    # --- Wire adapters ---
    def _get_service() -> ArticleService:
        session = SessionLocal()
        return ArticleService(
            repo=SqlAlchemyArticleRepository(session),
            search=ElasticsearchSearchIndex(es_client),
            notifications=EmailNotificationService(
                smtp_host="smtp.example.com",
                smtp_port=587,
                from_address="[email protected]",
                subscriber_lookup=lambda author_id: [],  # placeholder
            ),
            events=RedisEventBus(redis_client),
        )

    # Override the dependency
    app = FastAPI(title="Article Service")
    app.dependency_overrides[get_article_service] = _get_service
    app.include_router(router)

    return app


app = create_app()

:::tip The Composition Root Is the Only "Dirty" Place The composition root knows about every concrete class. That is by design. It is the single location where you choose which adapters to use. Every other module depends only on ports (Protocol interfaces). :::

Part 7 - Testing with Fake Adapters

The real payoff of hexagonal architecture is testing. You swap real adapters for in-memory fakes.

# tests/fakes.py
from typing import Optional
from uuid import UUID
from ports.outbound import Article


class FakeArticleRepository:
    def __init__(self) -> None:
        self._store: dict[UUID, Article] = {}

    def save(self, article: Article) -> None:
        self._store[article.id] = article

    def get_by_id(self, article_id: UUID) -> Optional[Article]:
        return self._store.get(article_id)

    def list_published(self, limit: int = 20, offset: int = 0) -> list[Article]:
        published = [a for a in self._store.values() if a.is_published]
        published.sort(key=lambda a: a.published_at or "", reverse=True)
        return published[offset : offset + limit]

    def delete(self, article_id: UUID) -> None:
        self._store.pop(article_id, None)


class FakeSearchIndex:
    def __init__(self) -> None:
        self._indexed: dict[UUID, Article] = {}

    def index_article(self, article: Article) -> None:
        self._indexed[article.id] = article

    def remove_article(self, article_id: UUID) -> None:
        self._indexed.pop(article_id, None)

    def search(self, query: str, limit: int = 10) -> list[UUID]:
        results = [
            a.id
            for a in self._indexed.values()
            if query.lower() in a.title.lower() or query.lower() in a.content.lower()
        ]
        return results[:limit]


class FakeNotificationService:
    def __init__(self) -> None:
        self.author_notifications: list[tuple[UUID, str]] = []
        self.subscriber_notifications: list[Article] = []

    def notify_author(self, author_id: UUID, message: str) -> None:
        self.author_notifications.append((author_id, message))

    def notify_subscribers(self, article: Article) -> None:
        self.subscriber_notifications.append(article)


class FakeEventBus:
    def __init__(self) -> None:
        self.events: list[tuple[str, dict]] = []

    def publish(self, event_type: str, payload: dict) -> None:
        self.events.append((event_type, payload))

# tests/test_article_service.py
import pytest
from uuid import uuid4
from core.article_service import ArticleService
from ports.inbound import CreateArticleCommand, PublishArticleCommand
from tests.fakes import (
    FakeArticleRepository,
    FakeEventBus,
    FakeNotificationService,
    FakeSearchIndex,
)


@pytest.fixture
def fakes():
    return {
        "repo": FakeArticleRepository(),
        "search": FakeSearchIndex(),
        "notifications": FakeNotificationService(),
        "events": FakeEventBus(),
    }


@pytest.fixture
def service(fakes):
    return ArticleService(**fakes)


def test_create_article(service, fakes):
    author_id = uuid4()
    result = service.create_article(
        CreateArticleCommand(title="Test", content="Body", author_id=author_id)
    )
    assert result.title == "Test"
    assert result.is_published is False
    assert len(fakes["events"].events) == 1
    assert fakes["events"].events[0][0] == "article.created"


def test_publish_article(service, fakes):
    author_id = uuid4()
    created = service.create_article(
        CreateArticleCommand(title="Test", content="Body", author_id=author_id)
    )
    published = service.publish_article(
        PublishArticleCommand(article_id=created.id, actor_id=author_id)
    )
    assert published.is_published is True
    assert len(fakes["search"]._indexed) == 1
    assert len(fakes["notifications"].subscriber_notifications) == 1


def test_publish_by_non_author_raises(service):
    author_id = uuid4()
    other_id = uuid4()
    created = service.create_article(
        CreateArticleCommand(title="Test", content="Body", author_id=author_id)
    )
    with pytest.raises(PermissionError):
        service.publish_article(
            PublishArticleCommand(article_id=created.id, actor_id=other_id)
        )


def test_publish_nonexistent_article_raises(service):
    with pytest.raises(ValueError, match="not found"):
        service.publish_article(
            PublishArticleCommand(article_id=uuid4(), actor_id=uuid4())
        )


def test_list_only_returns_published(service):
    author_id = uuid4()
    service.create_article(
        CreateArticleCommand(title="Draft", content="Body", author_id=author_id)
    )
    created = service.create_article(
        CreateArticleCommand(title="Published", content="Body", author_id=author_id)
    )
    service.publish_article(
        PublishArticleCommand(article_id=created.id, actor_id=author_id)
    )
    results = service.list_articles(20, 0)
    assert len(results) == 1
    assert results[0].title == "Published"

These tests execute in milliseconds. No database, no Elasticsearch, no Redis, no SMTP server.

Part 8 - Hexagonal vs Clean Architecture

Hexagonal and Clean Architecture solve the same fundamental problem - isolating business logic from infrastructure - but they emphasize different aspects.

Aspect	Clean Architecture	Hexagonal Architecture
Organizing metaphor	Concentric rings (layers)	Hexagon with pluggable sides
Layer count	Four explicit layers	Two sides (driving/driven) + core
Primary focus	Dependency rule (inward)	Symmetry of ports/adapters
Port terminology	Not explicit	Central concept
Use cases	Explicit layer	Part of application core
When to prefer	Complex domain logic with many layers	Systems with many external integrations

In practice, most Python projects blend ideas from both. The key principles are identical:

1. Business logic depends on nothing external
2. External systems are accessed through interfaces
3. Concrete wiring happens at the composition root

:::danger Common Mistake: Ports Without Purpose Do not create a port interface for every class. Ports exist at architectural boundaries - where your application meets the outside world. An internal helper class that formats strings does not need a port. :::

Part 9 - Project Structure for Hexagonal Architecture

article_platform/
├── ports/
│   ├── __init__.py
│   ├── inbound.py          # Primary port interfaces + commands
│   └── outbound.py         # Secondary port interfaces
├── core/
│   ├── __init__.py
│   ├── article_service.py  # Application core (implements primary ports)
│   └── domain.py           # Domain entities (if separated from ports)
├── adapters/
│   ├── __init__.py
│   ├── inbound/
│   │   ├── __init__.py
│   │   ├── fastapi_adapter.py
│   │   └── cli_adapter.py
│   └── outbound/
│       ├── __init__.py
│       ├── sqlalchemy_article_repo.py
│       ├── elasticsearch_search.py
│       ├── email_notifications.py
│       └── redis_events.py
├── tests/
│   ├── __init__.py
│   ├── fakes.py            # Fake adapters for testing
│   ├── test_article_service.py
│   ├── test_sqlalchemy_repo.py   # Integration test
│   └── test_fastapi_adapter.py   # API test
└── main.py                 # Composition root

The directory structure mirrors the architecture. Anyone new to the codebase can immediately see where ports, core logic, and adapters live.

Key Takeaways

• Ports define contracts, adapters fulfill them: ports are Protocol interfaces that describe what the application needs (secondary) and what it offers (primary). Adapters are concrete implementations.
• Primary adapters drive the application (REST, CLI, gRPC). Secondary adapters are driven by the application (database, email, search). The asymmetry matters for understanding data flow.
• The application core has zero framework imports: it depends only on domain entities and port interfaces defined with Protocol.
• The composition root is the single wiring point: it is the only module that imports concrete adapter classes and assembles the object graph.
• Fake adapters make testing trivial: in-memory implementations of secondary ports let you test all business logic without infrastructure, running tests in milliseconds.
• Hexagonal and Clean Architecture are complementary: blend the ideas that serve your project. The shared core principle - business logic depends on nothing external - is what matters.

Graded Practice Challenges

Level 1 - Identify the Pattern

Question 1: In the hexagonal model, is a REST API controller a primary adapter or a secondary adapter?

Answer

A REST API controller is a primary (driving) adapter. It drives the application by translating HTTP requests into commands that the application core understands. The adapter calls the application; the application does not call the adapter.

Question 2: Why does ArticleService accept SearchIndex as a constructor parameter instead of importing ElasticsearchSearchIndex directly?

Answer

Constructor injection with a Protocol type (SearchIndex) means ArticleService depends on an abstraction, not a concrete implementation. This allows swapping Elasticsearch for a different search backend (e.g., Meilisearch, OpenSearch, or a fake) without modifying any business logic. It also enables testing with FakeSearchIndex.

Question 3: What is wrong with this adapter?

class MyArticleRepo:
    def save(self, article: Article) -> None:
        article.publish()  # "convenience" - auto-publish on save
        self._session.merge(self._to_model(article))

Answer

The repository adapter is executing business logic (article.publish()). Adapters should only translate between the domain and the infrastructure. The decision to publish should be made in the application core (use case), not in the storage adapter. This adapter violates the single responsibility of adapters and makes the system harder to reason about.

Level 2 - Refactoring Challenge

You have a payment processing function that directly calls Stripe:

import stripe

def process_payment(user_id: int, amount: float, currency: str) -> str:
    stripe.api_key = "sk_test_xxx"
    customer = stripe.Customer.retrieve(f"cust_{user_id}")
    charge = stripe.Charge.create(
        amount=int(amount * 100),
        currency=currency,
        customer=customer.id,
    )
    return charge.id

Refactor this into hexagonal architecture: (a) define a PaymentGateway secondary port, (b) implement a StripePaymentAdapter, (c) create a ProcessPayment use case that depends on the port, (d) write a FakePaymentGateway for tests.

Level 3 - Design Challenge

Design the hexagonal architecture for an online examination system that supports:

• Multiple question types (MCQ, coding, essay)
• Timed exams with auto-submission
• Plagiarism detection via an external API
• Result computation with configurable grading rubrics
• PDF certificate generation

Identify all primary ports (what the system offers), all secondary ports (what the system needs), and draw the adapter map. Explain which adapters you would build as fakes first for rapid development.

What's Next

In the next lesson, Dependency Injection - Decoupling Components, we will dive into the mechanics of how dependencies are provided to your components - from manual constructor injection to DI containers and FastAPI's Depends system.