iktakahiro.dev

How to get started DDD & Onion-Architecture in Python web application

How to get started DDD & Onion-Architecture in Python web application

Takahiro Ikeuchi's photo
Takahiro Ikeuchi
·

6 min read

Featured on Hashnode

Table of contents

I published the following code for sharing knowledge of DDD & Onion Architecture in Python web applications.

https://github.com/iktakahiro/dddpy

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

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:

  1. Domain layer classes (such as an Entity or a Value Object) DO NOT extend the SQLAlchemy Base class.
  2. 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.

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.

1_gf-EW5sNS46NZcPxdVFBHw.jpeg

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.

PythonDDDsoftware architecturearchitecture