I published the following code for sharing knowledge of DDD & Onion Architecture in Python web applications.
iktakahiro
/
dddpy
Python DDD Example and Techniques
Python DDD & Onion-Architecture Example and Techniques
English | 日本語
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!
- My blog post: https://iktakahiro.dev/python-ddd-onion-architecture
Tech Stack
Code Architecture
The directory structure is based on Onion Architecture:
├── main.py
├── dddpy
│ ├── domain
│ │ └── book
│ ├── infrastructure
│ │ └── sqlite
│ │ ├── book
│ │ └── database.py
│ ├── presentation
│ │ └── schema
│ │ └── book
│ └── usecase
│ └── book
└── tests
Entity
To represent an Entity in Python, use the __eq__() method to ensure the object's identity.
class Book
def __init__(self, id: str, title: str):
self…I've also written a git of it on README. But in some ways, DDD is too difficult for us to understand; I would like to explain this architecture.
Motivation
My day job is to develop apps for smartphones using Flutter. In this development, I've adopted DDD with Onion architecture. This approach has worked well so far. On the other hand, the requirements of a native application are more complex than for typical web apps. So, I wanted to see what would happen if I applied DDD to a thin web app.
Conclusion
- DDD can be adopted in the Python world as well.
- For users familiar with MVC through prominent frameworks such as Django, this codebase may seem bizarre. But I believe it will help in many ways, especially to ensure maintainability and testability.
- The variety of interfaces that make up DDD is not necessarily compatible with the dynamically typed language. However, it is achievable using the standard features provided by Python.
Overview
Code Architecture
- DDD & Onion Architecture
├── main.py
├── dddpy
│ ├── domain
│ │ └── book
│ │ ├── book.py # Entity
│ │ ├── book_exception.py # Exception definitions
│ │ ├── book_repository.py # Repository interface
│ │ └── isbn.py # Value Object
│ ├── infrastructure
│ │ └── sqlite
│ │ ├── book
│ │ │ ├── book_dto.py # DTO using SQLAlchemy
│ │ │ ├── book_query_service.py # Query service implementation
│ │ │ └── book_repository.py # Repository implementation
│ │ └── database.py
│ ├── presentation
│ │ └── schema
│ │ └── book
│ │ └── book_error_message.py
│ └── usecase
│ └── book
│ ├── book_command_model.py # Write models including schemas of the RESTFul API
│ ├── book_command_usecase.py
│ ├── book_query_model.py # Read models including schemas
│ ├── book_query_service.py # Query service interface
│ └── book_query_usecase.py
└── tests
Tech Stack
FastAPI
FastAPI framework, high performance, easy to learn, fast to code, ready for production
fastapi.tiangolo.com
sqlalchemy.org
Techniques
- Repository pattern
- Data Transfer Object
- CQRS pattern
- Unit of Work pattern
- Dependency Injection (using FastAPI feature)
How to implement components of DDD
The assumption is that you are familiar with Eric Evans' book and Onion architecture, so I will not explain them. In the following, I will explain how you can implement each DDD component in the Python world.
Entity
To represent an Entity in Python, use __eq__() method to ensure the object's identity.
class Book:
def __init__(
self,
id: str,
title: str,
):
self.id: str = id
self.title: str = title
def __eq__(self, o: object) -> bool:
if isinstance(o, Book):
return self.id == o.id
return False
In this example, it's anemic. You have to implement more behaviors. I would add that the use of getter and setter is not Pythonic code. On the other hand, I think we can use it if necessary.
Value Object
As I understand it, the requirements for Value Object are the following:
- It's not an Entity
- If all the properties are the same, The objects are identical.
- As a result, they are interchangeable.
To represent a Value Object, use @dataclass decorator with eq=True and frozen=True.
@dataclass(init=False, eq=True, frozen=True)
class Isbn:
"""Isbn represents an ISBN code as a value object"""
value: str
def __init__(self, value: str):
if pattern.match(value) is None:
raise ValueError("isbn should be a valid format.")
object.__setattr__(self, "value", value)
Value Objects do not necessarily have to be immutable, but it is better to protect robustness.
Interface (abstract Class)
On a duck-typing language, IMO, it does not go well with defining interfaces. But Abstract Base class helps us to prepare them.
class BookRepository(ABC):
@abstractmethod
def create(self, book: Book) -> Optional[Book]:
raise NotImplementedError
@abstractmethod
def find_by_id(self, id: str) -> Optional[Book]:
raise NotImplementedError
Program to an interface, not an implementation.
Data Transfer Object and a factory method
DTO (Data Transfer Object) is a good practice to isolate domain objects from the infrastructure layer.
class BookDTO(Base):
__tablename__ = "book"
id: Union[str, Column] = Column(String, primary_key=True, autoincrement=False)
title: Union[str, Column] = Column(String, nullable=False)
On a minimum MVC architecture, models often inherit a base class provided by an O/R Mapper. But in that case, the domain layer would be dependent on the outer layer.
To solve this problem, we can set two rules:
- Domain layer classes (such as an Entity or a Value Object) DO NOT extend the SQLAlchemy Base class.
- Data transfer Objects extend the O/R mapper class.
Those, including my friend, who assert a DTO and a repository pattern are an overzealous approach.
I'm trying to tell you that you are not using the repository pattern to replace RDBs in the future. You have to take it to ensure testability today.
Read/Write models and request validation
It is helpful to separate a read-only model from a write-only model to implement the CQRS pattern.
I hadn't anticipated this, but this approach also worked well for managing input/output validation.
- https://github.com/iktakahiro/dddpy/blob/main/dddpy/usecase/book/book_query_model.py
- https://github.com/iktakahiro/dddpy/blob/main/dddpy/usecase/book/book_command_model.py
class BookReadModel(BaseModel):
id: str = Field(example="vytxeTZskVKR7C7WgdSP3d")
isbn: str = Field(example="978-0321125217")
FastAPI can set response_model and request_model. Amazingly, by associating the Read model with the Response model and the Write model with the Request model, API documentation can be generated and validated.
@app.post(
"/books",
response_model=BookReadModel,
status_code=status.HTTP_201_CREATED,
responses={
status.HTTP_409_CONFLICT: {
"model": ErrorMessageBookIsbnAlreadyExists,
},
},
)
async def create_book(
data: BookCreateModel,
book_command_usecase: BookCommandUseCase = Depends(book_command_usecase),
):
try:
book = book_command_usecase.create_book(data)
except BookIsbnAlreadyExistsError as e:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=e.message,
)
except Exception as e:
logger.error(e)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
return book
Amazingly, by associating read models with the response_model and write models with the request model, the framework can generate API documentation and validate the request body.
CQRS pattern
I will skip a general explanation of CQRS. Instead, I would like to show the specific sequence in the figure below.
- https://github.com/iktakahiro/dddpy/blob/main/dddpy/infrastructure/sqlite/book/book_query_service.py
- https://github.com/iktakahiro/dddpy/blob/main/dddpy/infrastructure/sqlite/book/book_repository.py
By splitting the Read model and Write model, we can flexibly respond to requests' input and output. My dear seniors educated me about DRY for a long time, but I've come to realize that it's not always absolute.
Unit of Work pattern
In this short project, the missing piece is transaction management. For the gap, the Unit of Work pattern fits almost entirely. But first, I would like to say that this is something that I would not strongly recommend to others, including you, because it is redundant, even for me, who suggested the repository pattern so much. However, I can't think of any other way to do transaction management without the assistance of a framework and its middleware.
class BookCommandUseCaseUnitOfWork(ABC):
book_repository: BookRepository
@abstractmethod
def begin(self):
raise NotImplementedError
@abstractmethod
def commit(self):
raise NotImplementedError
@abstractmethod
def rollback(self):
raise NotImplementedError
class BookCommandUseCaseUnitOfWorkImpl(BookCommandUseCaseUnitOfWork):
def __init__(
self,
session: Session,
book_repository: BookRepository,
):
self.session: Session = session
self.book_repository: BookRepository = book_repository
def begin(self):
self.session.begin()
def commit(self):
self.session.commit()
def rollback(self):
self.session.rollback()
Dependency Injection
This is a last topic, Dependency Injection sounds like an exaggeration, but it essentially means assigning a class instance to a property of other classes. One of the reasons I like FastAPI is that it provides a DI mechanism by default.
def book_query_usecase(session: Session = Depends(get_session)) -> BookQueryUseCase:
book_query_service: BookQueryService = BookQueryServiceImpl(session)
return BookQueryUseCaseImpl(book_query_service)
def book_command_usecase(session: Session = Depends(get_session)) -> BookCommandUseCase:
book_repository: BookRepository = BookRepositoryImpl(session)
uow: BookCommandUseCaseUnitOfWork = BookCommandUseCaseUnitOfWorkImpl(
session, book_repository=book_repository
)
return BookCommandUseCaseImpl(uow)
I don't think there is a widely known methodology for practicing DDD in Python. If you are going to try it, please refer to it.
Top comments (0)