💊

Django REST frameworkで戦術的DDDを実装してみた

2022/11/13に公開約9,700字

初めに

Django REST framework(DRF)は、Djangoを利用してWeb APIを簡単に実装できるライブラリです。
主な目的は個人的な勉強のためですが、少しでも誰かの参考になればと思ったのでDRFを用いて戦術的DDDパターンを実装したものをまとめました。
DDDについて解説をすることを目的としていないため、もしDDDを知らない方や勉強したい人は先に他の記事をご覧になった方が良いかと思います。

概要

今回はDjangoのチュートリアルの投票アプリをベースに一部処理を追加したシステムを作成しました。

ディレクトリ構成

先に今回作成したシステムのディレクトリ構成を紹介します。
コンテキスト単位でDjangoのアプリケーションを作成しており、各アプリケーションで共通で使用するものはshareに定義しています。

├── polls
│   ├── __init__.py
│   ├── dto
│   │   ├── __init__.py
│   │   └── questions_get_dto.py
│   ├── entities
│   │   ├── __init__.py
│   │   ├── choice_entity.py
│   │   └── question_entity.py
│   ├── factories
│   │   ├── __init__.py
│   │   ├── choice_factory.py
│   │   └── question_factory.py
│   ├── models.py
│   ├── query_services
│   │   ├── __init__.py
│   │   └── question_query_service.py
│   ├── repositories
│   │   ├── __init__.py
│   │   └── qustion_repository.py
│   ├── tests
│   ├── urls.py
│   ├── use_cases
│   │   ├── __init__.py
│   │   └── question_use_case.py
│   └── views
│       ├── __init__.py
│       ├── questions_choices_view.py
│       ├── questions_view.py
│       └── questions_vote_view.py
└── share
    ├── __init__.py
    └── base
        ├── __init__.py
	├── dto.py
        ├── entity.py
        └── model.py

model

基本的にはDjangoのチュートリアルのmodelをそのまま使用しています。

今回は識別子にUUIDを使用するため、UUIDをprimary_keyとするカスタマイズだけ行いました。
UUIDModelを継承してUUIDを識別子としたデータモデルを定義しています。

share/base/model.py
from django.db import models


class UUIDModel(models.Model):
    id = models.UUIDField(primary_key=True, editable=False)

    class Meta:
        abstract = True
polls/models.py
from django.db import models

from share.base.model import UUIDModel


class Question(UUIDModel):
    question_text = models.CharField(max_length=200)
    pub_date = models.DateTimeField('date published', auto_now_add=True)

    def __str__(self):
        return self.question_text


class Choice(UUIDModel):
    question = models.ForeignKey(Question, on_delete=models.CASCADE)
    choice_text = models.CharField(max_length=200)
    votes = models.IntegerField(default=0)

    def __str__(self):
        return self.choice_text

アーキテクチャ

今回はシンプルなレイヤードアーキテクチャを採用しました。

Presentation(views)

Application(UseCase)の実行し、APIとして外部との疎通を担当します。
django REST frameworkのAPIViewを用いて実装します。

今回はバリデーションやシリアライズは適当にしていますが、本来であればDjango標準のSerializerやdjango-rest-marshmallowのようなライブラリ等で行います。

polls/views/question_view.py
from rest_framework.views import APIView
from rest_framework.response import Response

from polls.use_cases.question_use_case import QuestionUseCase


class QuestionsView(APIView):
    def get(self, _):
        questions_get_dto = QuestionUseCase.get_latest_questions()
        return Response(questions_get_dto.as_dict(), status=200)

    def post(self, request):
        question_entity = QuestionUseCase.create_question(request.data['question_text'])
        return Response(question_entity.as_dict(), status=200)

Application(use_cases)

システムの振る舞いとしての業務ロジックを実装します。
基本的にはPresentationから呼び出され、EntityやRepository、QueryServiceなどを使用しながら業務ロジックを実現します。

新規登録の場合はFactoryを経由してEntityを作成してRepositoryに渡してDBに永続化します。
更新の場合はRepositoryからEntityを取得し、更新したEntityを再度Repositoryに渡すことでDBに反映させます。
単純にデータを取得する場合はQueryServiceからDTOを取得してViewに返します。

polls/use_cases/question_use_case.py
from polls.dto.questions_get_dto import QuestionsGetDto
from polls.entities.question_entity import QuestionEntity
from polls.factories.choice_factory import ChoiceFactory
from polls.factories.question_factory import QuestionFactory
from polls.repositories.qustion_repository import QuestionRepository
from polls.query_services.question_query_service import QuestionQueryService


class QuestionUseCase:
    @classmethod
    def get_by_id(cls, question_id: str) -> QuestionEntity:
        return QuestionRepository.get_by_id(question_id)

    @classmethod
    def get_latest_questions(cls) -> QuestionsGetDto:
        return QuestionQueryService.get_latest_questions()

    @classmethod
    def create_question(cls, question_text: str) -> QuestionEntity:
        question_entity = QuestionFactory.build(question_text)
        return QuestionRepository.save(question_entity)

    @classmethod
    def create_choice(cls, question_id: str, choice_text: str) -> QuestionEntity:
        question_entity = QuestionRepository.get_by_id(question_id)
        choice_entity = ChoiceFactory.build(question_id, choice_text)
        question_entity.add_choice(choice_entity)
        return QuestionRepository.save(question_entity)

    @classmethod
    def vote(cls, question_id: str, choice_id: str) -> QuestionEntity:
        question_entity = QuestionRepository.get_by_id(question_id)
        question_entity.vote(choice_id)
        return QuestionRepository.save(question_entity)

Domain(entities)

ソフトウェアとしてのドメインルールを表現しています。
Pythonの標準ライブラリであるdataclassやuuidを使用しています。
識別子であるIDの早期生成の仕組みはこちらの記事を参考にしました。

ちなみにas_dict関数はviewでのレスポンスでシリアライズするために作成しています。
Presentationでも書いた通り、本来は何かしらのシリアライズの仕組みを使用した方が良いです。

share/base/entity.py
from dataclasses import asdict, dataclass, field
import uuid


@dataclass
class Entity:
    id: uuid.UUID = field(default_factory=lambda: globals()['Entity'].next_id(), kw_only=True)

    @classmethod
    def next_id(cls) -> uuid.uuid4:
        return uuid.uuid4()

    def as_dict(self):
        return asdict(self)
polls/entities/question_entity.py
from dataclasses import dataclass
from typing import List

from share.base.entity import Entity
from polls.entities.choice_entity import ChoiceEntity


@dataclass
class QuestionEntity(Entity):
    question_text: str
    pub_date: str
    choice_entities: List[ChoiceEntity]

    def vote(self, choice_id: str):
        choice_entity = self.get_choice_entity(choice_id)
        choice_entity.vote()

    def get_choice_entity(self, choice_id: str) -> ChoiceEntity:
        return next(filter(lambda c: str(c.id) == choice_id, self.choice_entities))

    def add_choice(self, choice_entity: ChoiceEntity):
        self.choice_entities.append(choice_entity)

Factory

Entityを新規作成する場合や複雑なEntityを生成する場合に使用します。

polls/factories/question_factory.py
from django.utils import timezone

from polls.entities.question_entity import QuestionEntity


class QuestionFactory:
    @classmethod
    def build(cls, question_text: str) -> QuestionEntity:
        return QuestionEntity(question_text, str(timezone.now()), [])

Infrastructure(repositories, query_services)

DjangoのActiveRecordを使用してデータの永続化や検索を実装しています。

今回の実装ではコマンドクエリ責務分離(CQRS)を採用しています。
CQRSはデータの参照と更新の責任を分離するパターンです。
CQRSの思想はさまざまありますが、イベントソーシングとの併用やデータベースレベルで分離する必要はなく、参照と更新の責任のモデルを分離すればOKとしています。

今回の実装では、更新系モデルをEntity、参照系モデルをDTOとしています。
Repositoryは更新系モデルのEntityの永続化、検索を担当し、参照系モデルであるDTOはQueryServiceから取得します。
どちらもActiveRecordを通してデータベースとやりとりするため、Infrastructure層に該当いたします。

CQRSについての詳細はこちらの記事をご覧ください。
https://little-hands.hatenablog.com/entry/2019/12/02/cqrs

Repository

polls/repositories/question_repository.py
from polls.models import Choice, Question
from polls.entities.choice_entity import ChoiceEntity
from polls.entities.question_entity import QuestionEntity


class QuestionRepository:
    @classmethod
    def get_by_id(cls, question_id: str) -> QuestionEntity:
        question = Question.objects.prefetch_related('choice_set').get(pk=question_id)
        choice_entities = [
            ChoiceEntity(c.choice_text, c.votes, question.id, id=c.id) for c in question.choice_set.all()
        ]
        return QuestionEntity(question.question_text, question.pub_date, choice_entities, id=question.id)

    @classmethod
    def save(cls, question_entity: QuestionEntity) -> QuestionEntity:
        question, _ = Question.objects.get_or_create(pk=question_entity.id)
        question.question_text = question_entity.question_text
        question.pub_date = question_entity.pub_date
        question.save()

        for choice_entity in question_entity.choice_entities:
            choice, _ = Choice.objects.get_or_create(pk=choice_entity.id, question=question)
            choice.choice_text = choice_entity.choice_text
            choice.votes = choice_entity.votes
            choice.save()

        return question_entity

QueryService

polls/query_services/qustion_query_service.py
from polls.dto.questions_get_dto import QuestionsGetDto, QuestionDto
from polls.models import Question


class QuestionQueryService:
    @classmethod
    def get_latest_questions(cls, limit: int = 5) -> QuestionsGetDto:
        latest_question_list = Question.objects.order_by('-pub_date')[:limit]
        question_list = [QuestionDto(q.id, q.question_text, q.pub_date) for q in latest_question_list]
        return QuestionsGetDto(question_list)

DTOはdataclassを用いてシンプルに定義しています。

share/base/dto.py
from dataclasses import asdict, dataclass


@dataclass
class Dto:
    def as_dict(self):
        return asdict(self)
polls/dto/question_get_dto.py
from dataclasses import dataclass
from typing import List

from share.base.dto import Dto


@dataclass
class QuestionDto(Dto):
    id: int
    question_text: str
    pub_date: str


@dataclass
class QuestionsGetDto(Dto):
    questions: List[QuestionDto]

まとめ

DRFを用いて戦術的DDDパターンを実装してみました。
modelやentityの識別子など一部工夫をしましたが、思っていたよりもシンプルに実装できました。

Djangoはアプリケーションの単位でコンテキストを区切りやすいと感じたので、小規模アプリケーションや個人開発アプリなどでDjango×DDDで作ってみたいと思います。

最後に

今回作成したソースコードを載せておきます。
ちなみにdjango-environやflake8などDjangoで開発する際によく使用されるであろうライブラリもいくつか設定しています。よろしければぜひテンプレートとして使ってみてください。
https://github.com/kenta-takeuchi/django-ddd-example

Discussion

ログインするとコメントできます