Django REST frameworkで戦術的DDDを実装してみた
初めに
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を識別子としたデータモデルを定義しています。
from django.db import models
class UUIDModel(models.Model):
id = models.UUIDField(primary_key=True, editable=False)
class Meta:
abstract = True
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のようなライブラリ等で行います。
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に返します。
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でも書いた通り、本来は何かしらのシリアライズの仕組みを使用した方が良いです。
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)
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を生成する場合に使用します。
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についての詳細はこちらの記事をご覧ください。
Repository
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
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を用いてシンプルに定義しています。
from dataclasses import asdict, dataclass
@dataclass
class Dto:
def as_dict(self):
return asdict(self)
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で開発する際によく使用されるであろうライブラリもいくつか設定しています。よろしければぜひテンプレートとして使ってみてください。
Discussion