Skip to content

DDD Service — Usage Examples

Basic examples demonstrating the DDD hexagonal architecture in practice.

See also: Core concepts · External API calls · Bank balance alert


Minimal end-to-end demo (in-memory)

A quick example wiring all layers together:

from modules.example_feature.infrastructure.repositories import InMemoryNoteRepository
from modules.example_feature.application.use_cases import CreateNote, ListNotes

repo = InMemoryNoteRepository()
create_note = CreateNote(repo)
list_notes = ListNotes(repo)

create_note.execute("First note")
print(list_notes.execute())

Wiring with FastAPI

Example composition and HTTP handler:

# composition.py
from modules.example_feature.infrastructure.repositories import InMemoryNoteRepository
from modules.example_feature.application.use_cases import CreateNote, ListNotes

repo = InMemoryNoteRepository()
create_note = CreateNote(repo)
list_notes = ListNotes(repo)
# api/notes.py
from fastapi import APIRouter
from pydantic import BaseModel
from .composition import create_note, list_notes

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


class CreateNotePayload(BaseModel):
    title: str


@router.post("/")
def create_note_endpoint(payload: CreateNotePayload):
    note = create_note.execute(title=payload.title)
    return {
        "id": note.id,
        "title": note.title,
        "created_at": note.created_at.isoformat(),
    }


@router.get("/")
def list_notes_endpoint():
    return [
        {"id": n.id, "title": n.title, "created_at": n.created_at.isoformat()}
        for n in list_notes.execute()
    ]

Swapping implementations

The power of ports-and-adapters: swap the repository without touching use-cases.

# For tests — use in-memory
from modules.example_feature.infrastructure.repositories import InMemoryNoteRepository
repo = InMemoryNoteRepository()

# For production — use PostgreSQL (same interface)
from modules.example_feature.infrastructure.postgres_repository import PostgresNoteRepository
repo = PostgresNoteRepository(connection_string="postgresql://...")

# Use-case stays the same
create_note = CreateNote(repo)

Testing with mocks

import pytest
from unittest.mock import Mock
from modules.example_feature.application.use_cases import CreateNote
from modules.example_feature.domain.ports import NoteRepository


def test_create_note():
    mock_repo = Mock(spec=NoteRepository)
    mock_repo.add.return_value = Mock(id="123", title="Test", created_at="2026-01-01")

    use_case = CreateNote(mock_repo)
    result = use_case.execute("Test")

    mock_repo.add.assert_called_once()
    assert result.id == "123"