Python DDD & Onion-Architecture Example and Techniques

English | 日本語

• Go implementation: oniongo

NOTE: This repository is an example to demonstrate "how to implement DDD architecture in a Python web application." If you use this as a reference, ensure to implement authentication and security before deploying it to a real-world environment!

• DeepWiki powered by Devin:
• My blog post:

Tech Stack

• FastAPI
• SQLAlchemy

* SQLite

• uv
• Docker

Prerequisites

• Python 3.13 or higher
• uv - Python package installer and resolver

Project Setup

• Install dependencies using uv:

make install

• Run the web app

make dev

Code Architecture

The directory structure is based on Onion Architecture:

├── main.py
├── dddpy
│   ├── domain
│   │   └── todo
│   │       ├── entities
│   │       │   └── todo.py
│   │       ├── value_objects
│   │       │   ├── todo_title.py
│   │       │   ├── todo_description.py
│   │       │   ├── todo_id.py
│   │       │   └── todo_status.py
│   │       ├── repositories
│   │       │   └── todo_repository.py
│   │       └── exceptions
│   ├── infrastructure
│   │   ├── di
│   │   │   └── injection.py
│   │   └── sqlite
│   │       ├── database.py
│   │       └── todo
│   │           ├── todo_repository.py
│   │           └── todo_dto.py
│   ├── presentation
│   │   └── api
│   │       └── todo
│   │           ├── handlers
│   │           │   └── todoapiroute_handler.py
│   │           ├── schemas
│   │           │   └── todo_schema.py
│   │           └── error_messages
│   │               └── todoerrormessage.py
│   └── usecase
│       └── todo
│           ├── createtodousecase.py
│           ├── updatetodousecase.py
│           ├── starttodousecase.py
│           ├── findtodosusecase.py
│           ├── findtodobyidusecase.py
│           ├── completetodousecase.py
│           └── deletetodousecase.py
└── tests

Domain Layer

The domain layer contains the core business logic and rules. It includes:

• Entities
• Value Objects
• Repository Interfaces

Here's how each component is implemented in this project:

1. Entities

Entities are domain models with unique identifiers. In this project, the Todo class is implemented as an entity:

class Todo:
    def init(
        self,
        id: TodoId,
        title: TodoTitle,
        description: Optional[TodoDescription] = None,
        status: TodoStatus = TodoStatus.NOT_STARTED,
        created_at: datetime = datetime.now(),
        updated_at: datetime = datetime.now(),
        completed_at: Optional[datetime] = None,
    ):
        self._id = id
        self._title = title
        self._description = description
        self._status = status
        self.createdat = created_at
        self.updatedat = updated_at
        self.completedat = completed_at
def eq(self, obj: object) -> bool:         if isinstance(obj, Todo):             return self.id == obj.id         return False

Key characteristics of entities:

• Have a unique identifier (id)
• Can change state (e.g., updatetitle, updatedescription, start, complete methods)
• Identity is determined by the identifier (eq method implementation)
• May be created through factory methods (e.g., create)

In this project, the eq method is implemented to determine instance identity solely based on the id:

def eq(self, obj: object) -> bool:
    if isinstance(obj, Todo):
        return self.id == obj.id # Note: Accessing via property
    return False

Key points of this implementation:

• Identity is determined solely by the identifier (id)
• Type safety is ensured with isinstance check

2. Value Objects

Value objects are immutable domain models without identifiers. This project implements several value objects:

@dataclass(frozen=True)
class TodoTitle:
    value: str
def post_init(self):         if not self.value:             raise ValueError('Title is required')         if len(self.value) > 100:             raise ValueError('Title must be 100 characters or less')

Key characteristics of value objects:

• Immutability guaranteed by @dataclass(frozen=True)
• Include value validation logic (post_init)
• No identifier
• Identity is determined by value content

3. Repository Interfaces

Repositories are abstraction layers responsible for entity persistence. This project implements the TodoRepository interface:

class TodoRepository(ABC):
    @abstractmethod
    def save(self, todo: Todo) -> None:
        """Save a Todo"""
@abstractmethod     def findbyid(self, todo_id: TodoId) -> Optional[Todo]:         """Find a Todo by ID"""
@abstractmethod     def find_all(self) -> List[Todo]:         """Get all Todos"""
@abstractmethod     def delete(self, todo_id: TodoId) -> None:         """Delete a Todo by ID"""

Key characteristics of repositories:

• Abstract entity persistence
• Define boundaries between domain and infrastructure layers
• Concrete implementations provided in the infrastructure layer

Infrastructure Layer

The infrastructure layer implements the interfaces defined in the domain layer. It includes:

• Database configurations
• Repository implementations
• External service integrations
• Dependency Injection (DI) setup

This layer provides concrete implementations of the abstractions defined in the domain layer, enabling the application to interact with external resources while maintaining clean separation of concerns.

1. Repository Implementations

The repository implementation is done as follows:

class TodoRepositoryImpl(TodoRepository):
    """SQLite implementation of Todo repository interface."""
def init(self, session: Session):         """Initialize repository with SQLAlchemy session."""         self.session = session
def findbyid(self, todo_id: TodoId) -> Optional[Todo]:         """Find a Todo by its ID."""         try:             row = self.session.query(TodoDTO).filterby(id=todoid.value).one()         except NoResultFound:             return None
return row.to_entity()
def save(self, todo: Todo) -> None:         """Save a new Todo item."""         tododto = TodoDTO.fromentity(todo)         try:             existing_todo = (                 self.session.query(TodoDTO).filter_by(id=todo.id.value).one()             )         except NoResultFound:             self.session.add(todo_dto)
else:             existingtodo.title = tododto.title             existingtodo.description = tododto.description             existingtodo.status = tododto.status             existingtodo.updatedat = tododto.updatedat             existingtodo.completedat = tododto.completedat

Unlike the repository interface, the implementation code in the infrastructure layer can contain details specific to a particular technology (SQLite in this example). It's often beneficial to clearly indicate the underlying technology in directory names (e.g., sqlite) or class names, rather than strictly adhering to abstract interface definitions.

2. Data Transfer Object

In Onion Architecture, inner layers (like the domain layer) do not depend on outer layers (infrastructure, presentation). Therefore, when exchanging data between layers, object conversion might be necessary to prevent details of one layer (e.g., infrastructure's database model) from leaking into others. Data Transfer Objects (DTOs) fulfill this conversion role. DTOs are simple objects used to transfer data across layers.

The TodoDTO class example below is an SQLAlchemy model (inheriting from Base) and includes methods (toentity, fromentity) for converting between itself and the domain entity (Todo):

class TodoDTO(Base):
    """Data Transfer Object for Todo entity in SQLite database."""
tablename = 'todo'     id: Mapped[UUID] = mappedcolumn(primarykey=True, autoincrement=False)     title: Mapped[str] = mapped_column(String(100), nullable=False)     description: Mapped[str] = mapped_column(String(1000), nullable=True)     status: Mapped[str] = mapped_column(index=True, nullable=False)     createdat: Mapped[int] = mappedcolumn(index=True, nullable=False)     updatedat: Mapped[int] = mappedcolumn(index=True, nullable=False)     completedat: Mapped[int] = mappedcolumn(index=True, nullable=True)
def to_entity(self) -> Todo:         """Convert DTO to domain entity."""         return Todo(             TodoId(self.id),             TodoTitle(self.title),             TodoDescription(self.description),             TodoStatus(self.status),             datetime.fromtimestamp(self.created_at / 1000, tz=timezone.utc),             datetime.fromtimestamp(self.updated_at / 1000, tz=timezone.utc),             datetime.fromtimestamp(self.completed_at / 1000, tz=timezone.utc)             if self.completed_at             else None,         )
@staticmethod     def from_entity(todo: Todo) -> 'TodoDTO':         """Convert domain entity to DTO."""         return TodoDTO(             id=todo.id.value,             title=todo.title.value,             description=todo.description.value if todo.description else None,             status=todo.status.value,             createdat=int(todo.createdat.timestamp() * 1000),             updatedat=int(todo.updatedat.timestamp() * 1000),             completedat=int(todo.completedat.timestamp() * 1000)             if todo.completed_at             else None,         )

By converting TodoDTO objects (dependent on SQLAlchemy) retrieved from the database into the domain layer's Todo entity before returning them to the use case layer, we prevent the use case layer from depending on infrastructure layer details. This also maintains consistency with the return type (Todo entity) defined in the repository interface.

3. Dependency Injection

This project uses FastAPI's dependency injection system to manage dependencies across layers. The DI configuration is centralized in the infrastructure/di/injection.py module:

def get_session() -> Iterator[Session]:
    """Yield a managed SQLAlchemy session for request handling."""
    session: Session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()
def gettodorepository(session: Session = Depends(get_session)) -> TodoRepository:     """Provide a repository instance bound to the current session."""     return newtodorepository(session)
def getcreatetodo_usecase(     todorepository: TodoRepository = Depends(gettodo_repository), ) -> CreateTodoUseCase:     """Provide the create-todo use case with injected repository."""     return newcreatetodousecase(todorepository)

Key benefits of this approach:

• Lifecycle Management: Database sessions are automatically managed (commit/rollback/close)
• Testability: Dependencies can be easily mocked or replaced in tests
• Loose Coupling: Presentation layer depends only on use case interfaces, not implementations
• Single Responsibility: Each dependency provider has one clear purpose

Usecase Layer

The usecase layer contains the application-specific business rules. It includes:

• Usecase implementations
• Error handling related to use cases

In this project, each use case is implemented as a separate class with a single public execute method, following the rule of "one public method per use case". This design ensures clear separation of concerns and makes the code more maintainable. Here's how it's implemented:

1. Use Case Interface and Implementation

Each use case follows this structure:

class CreateTodoUseCase:
    """CreateTodoUseCase defines a use case interface for creating a new Todo."""
@abstractmethod     def execute(         self, title: TodoTitle, description: Optional[TodoDescription] = None     ) -> Todo:         """execute creates a new Todo."""
class CreateTodoUseCaseImpl(CreateTodoUseCase):     """CreateTodoUseCaseImpl implements the use case for creating a new Todo."""
def init(self, todo_repository: TodoRepository):         self.todorepository = todorepository
def execute(         self, title: TodoTitle, description: Optional[TodoDescription] = None     ) -> Todo:         """execute creates a new Todo."""         todo = Todo.create(title=title, description=description)         self.todo_repository.save(todo)         return todo

Key characteristics of use cases:

• One class per use case
• Single responsibility principle
• Clear interface definition
• Dependency injection through constructor
• Factory function for instantiation

2. Error Handling

Use cases handle domain-specific errors:

class StartTodoUseCaseImpl(StartTodoUseCase):
    # ... init ...
def execute(self, todo_id: TodoId) -> Todo: # Corrected return type         todo = self.todorepository.findbyid(todoid)
if todo is None:             raise TodoNotFoundError
if todo.is_completed:             raise TodoAlreadyCompletedError
if todo.status == TodoStatus.IN_PROGRESS:             raise TodoAlreadyStartedError
todo.start()         self.todo_repository.save(todo)         return todo # Return the updated Todo

Presentation Layer

The presentation layer handles HTTP requests and responses. It includes:

• FastAPI route handlers
• Request/Response models
• Input validation

The handlers are organized under the presentation/api directory, which represents the API layer of the application. Each domain (like todo) has its own controller, schema, and error message definitions.

How to Use

Option 1: Local Development

• Ensure Python 3.13+ and uv are installed
• Clone this repository
• Run make install to install dependencies
• Run make dev to start the development server
• Access the API documentation at http://127.0.0.1:8000/docs

Option 2: Dev Container (VSCode)

• Clone and open this repository using VSCode
• Install the Dev Containers extension
• Open the command palette (Cmd/Ctrl+Shift+P) and select "Dev Containers: Reopen in Container"
• Once the container is built, run make dev in the terminal
• Access the API documentation at http://127.0.0.1:8000/docs

Sample Requests for the RESTful API

• Create a new todo:

curl --location --request POST 'localhost:8000/todos' \
--header 'Content-Type: application/json' \
--data-raw '{
    "title": "Implement DDD architecture",
    "description": "Create a sample application using DDD principles"
}'

• Response of the POST request:

{
    "id": "550e8400-e29b-41d4-a716-446655440000",
    "title": "Implement DDD architecture",
    "description": "Create a sample application using DDD principles",
    "status": "not_started", # Corrected status
    "created_at": 1614007224642,
    "updated_at": 1614007224642
}

• Get todos:

curl --location --request GET 'localhost:8000/todos'

• Response of the GET request:

[
    {
        "id": "550e8400-e29b-41d4-a716-446655440000",
        "title": "Implement DDD architecture",
        "description": "Create a sample application using DDD principles",
        "status": "not_started",
        "created_at": 1614007224642,
        "updated_at": 1614007224642
    }
]

• Start a todo:

curl --location --request PATCH 'localhost:8000/todos/550e8400-e29b-41d4-a716-446655440000/start'

• Complete a todo:

curl --location --request PATCH 'localhost:8000/todos/550e8400-e29b-41d4-a716-446655440000/complete'

• Update a todo:

curl --location --request PUT 'localhost:8000/todos/550e8400-e29b-41d4-a716-446655440000' \
--header 'Content-Type: application/json' \
--data-raw '{
    "title": "Updated title",
    "description": "Updated description"
}'

Development

Running Tests

make test

This command runs both type checking with Pyrefly and unit tests with pytest.

Code Quality

This project uses several tools to maintain code quality:

• Pyrefly for static type checking
• ruff for linting & formatting
• pytest for testing

Code Formatting

Format your code using ruff:

make format

Docker Development

The project includes a .devcontainer configuration for Docker-based development. This ensures a consistent development environment across different machines. See How to Use for setup instructions.

License

This project is licensed under the MIT License - see the LICENSE file for details.