Domain driven design in Python 🐍

A very simple application

Client

I need a todo list app

class Task( models.Model ):
	completed = models.BooleanField( default = False )
	title = models.CharField( max_length = 100 )

Django!

Client

OK, now send the user an email anytime he completes a task

Architecture

Django

mailer_service.py

models.py

signals.py

Business logic

@receiver(pre_save, sender=Task)
def send_mail_on_complete(sender, **kwargs):
    if sender.completed:
        mailer_service.send(...)

Test?

  • Patch mailer_service
  • Trigger the signal with a completed task
  • Assert the mailer_service was called

Problems

⁉️ Stub any framework component

⁉️ Hard to test framework-linked functions

⁉️ Hard to guess where the business logic is

Client

Why is it so long to develop? It's easy to send a mail, right?

DDD

Domain-driven design is an approach to software development for complex needs by connecting the implementation to an evolving model.

Separate the model from the implementation

└── domain
    ├── task
    |   ├── entity
    |   |   └── task.py
    |   ├── repository
    |   |   └── task_repository.py
    |   └── service
    |       └── task_service.py
    └── mail
        └── service
            └── mail_service.py

Task entity

# domain/task/entity/task.py

class Task():
    def __init__(self, title : str):
        self.id = uuid4()
        self.title = title
        self.completed = False

└── domain
    ├── task
    |   ├── entity
    |   |   └── task.py
    |   ├── repository
    |   |   └── task_repository.py
    |   └── service
    |       └── task_service.py
    └── mail
        └── service
            └── mail_service.py

Task repository

# domain/task/repository/task_repository.py
import abc


class TaskRepository(abc.ABC):

    @abstractmethod
    def find(self, id : UUID):
        pass

    @abstractmethod
    def save(self, task : Task):
        pass
└── domain
    ├── task
    |   ├── entity
    |   |   └── task.py
    |   ├── repository
    |   |   └── task_repository.py
    |   └── service
    |       └── task_service.py
    └── mail
        └── service
            └── mail_service.py

Task service

# domain/task/service/task_service.py


class TaskService():
    def __init__(self, task_repository : TaskRepository, mailer_service : MailerService):
        self.task_repository = task_repository
        self.mailer_service = mailer_service

    def complete_task(self, task_id : UUID) -> None:
        task = self.task_repository.find(task_id)

        assert not task.completed, 'Task {} already completed'.format(task)

        task.completed = True
        self.mailer_service.send(...)

        self.task_repository.save(task)
└── domain
    ├── task
    |   ├── entity
    |   |   └── task.py
    |   ├── repository
    |   |   └── task_repository.py
    |   └── service
    |       └── task_service.py
    └── mail
        └── service
            └── mail_service.py

Mailer service

# domain/mail/service/mailer_service.py


class MailerService(abc.ABC):

    @abstractmethod
    def send(self, mail : Mail):
        pass

└── domain
    ├── task
    |   ├── entity
    |   |   └── task.py
    |   ├── repository
    |   |   └── task_repository.py
    |   └── service
    |       └── task_service.py
    └── mail
        └── service
            └── mail_service.py

Tests?

Dependency injection!

# domain/task/service/task_service.py


class TaskService():
    def __init__(self, task_repository : TaskRepository, mailer_service : MailerService):
        self.task_repository = task_repository
        self.mailer_service = mailer_service

    def complete_task(self, task_id : UUID) -> None:
        task = self.task_repository.find(task_id)

        assert not task.completed, 'Task {} already completed'.format(task)

        task.completed = True
        self.mailer_service.send(...)

        self.task_repository.save(task)

Linking the domain with the infrastructure

Dependency injection

  1. Write concrete implementations for the domain interfaces, using the infrastructure
  2. Inject these implementations in the domain classes

1.Write concrete implementations for the domain interfaces, using the infrastructure

# infrastructure/repository/task_repository.py
from domain.task.repository.task_repository import TaskRepository as TaskRepositoryInterface
from infrastructure.models import TaskModel


class TaskRepository(TaskRepositoryInterface):
    def find(self, id : UUID):
        return TaskModel.objects.find(pk=id)

    def save(self, task : Task):
        persisted_task = self.find(task.id)
        persisted_task.completed = task.completed
        persisted_task.title = task.title

        persisted_task.save()

2.Inject these implementations in the domain classes

# infrastructre/service/task_service

from domain.task.service.task_service import TaskService
from infrastructure.service.mailer_service import MailerService
from infrastructure.repository.task_repository import TaskRepository


task_service = TaskService(TaskRepository(), MailerService())

Use libs!

Python-inject

# infrastructure/injector.py

import inject
from infrastructure.repository.task_repository import TaskRepository
from domain.task.repository.task_repository import TaskRepository as TaskRepositoryInterface

def my_config(binder):
    binder.bind(TaskRepositoryInterface, TaskRepository())

# Configure a shared injector.
inject.configure(my_config)
# domain/task/service/task_service.py

import inject
from domain.task.repository.task_repository import TaskRepository
from domain.mail.service.mailer_service import MailerService


class TaskService():
    task_repository = inject.attr(TaskRepository)
    mailer_service = inject.attr(MailerService)

    # ...

Victory!

⁉️ Stub any framework component

⁉️ Hard to test framework-linked functions

⁉️ Hard to guess where the business logic is

✅ Every class domain is easy to test

✅ The business logic is located in the domain

✅ Domain code is understandable by domain experts

Drawbacks

  • More code to write
  • Confusing for junior developers
  • More abstractions

Questions?