DDD(ドメイン駆動開発)をFastAPIで実装するときのプラクティスができたので、共有になります。
一応動いていますけど、もっといい実装方法があれば教えて下さい。そんな気持ちです。
この記事で分かること
-
この記事で分かること
- FastAPI+DDD のプラクティス（具体的には ↓）
- FastAPI における DDD を取り入れたディレクトリ構成
- FastAPI の Model と DDD の Value Object のデータ扱い方(データ変換)
- テストのための InMemoryRepository の実装
- FastAPI から自動生成される swagger と DDD
- おまけ) pycache のディレクトリが散在する
- FastAPI+DDD のプラクティス（具体的には ↓）
-
この記事で分からないこと
- FastAPI のゼロからの開発手順
- DDD のゼロからの知識を学ぶ
前提
言語、ミドルウェアのバージョン
| 言語、ミドルウェア | バージョン |
|---|---|
| python | 3.7.13 |
| FastAPI | 0.75.1 |
| sqlalchemy | 1.4.35 |
例題: アンケートできるサービス
ここで扱う題材として、QA が投稿できるサービスを想定しています。
特にこの記事では Question（質問）する部分についてフォーカスして書いていきます。
オニオンアーキテクチャ
コードの書き方としては、オニオンアーキテクチャをなんとなく参考にしている感じです。
DDD の参考書籍
今回のDDDの実践については、この書籍を参考にしています。
ポイント1: FastAPI における DDD を取り入れたディレクトリ構成
FastAPI/
|- src/
|- database
|- ddd
|- domain
|- application_service
|- repository
|- service
|- value_object
|- migration
|- routers
|- tests
databaseディレクトリについて
databaseには FastAPI としての Model が格納されます。DDD でいうとDataModelに相当します。
domainディレクトリについて
domainにはapplication_service、repository、service、value_objectといった各レイヤーのディレクトリを配置しました。
（もしかすると他にもいいディレクトリ構造があるかもしれませんけども、DDD 的な情報の整理はできるので今回は OK としました。）
migrationディレクトリについて
migrationにはマイグレーションのパッケージであるalembicを利用しました。
routersディレクトリについて
routersは FastAPI における各 API のルートを担当します。
testsディレクトリについて
testsにはdomain、つまり DDD 的な意味での各レイヤーのテストが格納されます。
今回はこのプロジェクトではテストツールとしてpytestを利用しました。
以下にそれぞれのレイヤーの実装したコードとテストコードを示します。
value_object のテストの例
def test_question_create(self):
question_id = fake.pyint()
question_text = fake.sentence()
question = Question(
QuestionId(question_id),
QuestionText(question_text)
)
assert question.id.get_id() == QuestionId(question_id).get_id()
assert question.question_text.get_text() == QuestionText(question_text).get_text()
repository のテストの例
class TestSqlAlchemyQuestionRepository:
def test_save(self):
fake = Faker()
question_id = fake.pyint()
question_text = fake.sentence()
question = Question(
QuestionId(question_id),
QuestionText(question_text)
)
self.repository = SqlAlchemyQuestionRepository(ENGINE)
self.repository.save(question)
subject = self.repository.find(question.id)
assert question.id.get_id() == subject.id.get_id()
assert question.application_id.get_id() == subject.application_id.get_id()
assert question.application_code.get_code() == subject.application_code.get_code()
assert question.question_text.get_text() == subject.question_text.get_text()
service のテストの例
都合により割愛。
application_service のテストの例
fixture を使うと便利です。
@pytest.fixture
def question_application_service(self):
self.question_repository = InMemoryQuestionRepository()
self.application_repository = InMemoryApplicationRepository()
self.question_application_service = QuestionApplicationService(
self.question_repository,
self.application_repository
)
def test_register_question(self, question_application_service):
fake = Faker()
question_text = fake.sentence()
question_update_command = QuestionUpdateCommand(
id=None,
question_text=question_text
)
registered_questions = self.question_application_service.register_question(question_update_command)
subject = registered_questions[0]
assert isinstance(subject.id.get_id(), int) == True
assert question_text == subject.question_text.get_text()
ポイント2: FastAPI の Model と DDD の Value Object のデータ扱い方(データ変換)
FastAPI にも Model という概念がありますが、DDD にも Model という言葉を使うので、
以下のように整理しておきます。
| 整理前 | 整理後 | |
|---|---|---|
| FastAPI | Model | DataModel |
| DDD | Value Object | Model |
from ...value_object.question.question import Question
from ...value_object.question.question_id import QuestionId
from ...value_object.question.question_text import QuestionText
from .i_question_repository import IQuestionRepository
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import func
from sqlalchemy.dialects.postgresql import insert
import pdb
import sys
import os
sys.path.append(os.path.join(os.path.dirname(__file__), '../../../'))
from database.setting.setting import ENGINE, Base
from database.model.question import QuestionDataModel
class SqlAlchemyQuestionRepository(IQuestionRepository):
def __init__(self, ENGINE):
self.session = sessionmaker(ENGINE)
def toModel(self, question_datamodel:QuestionDataModel):
question_id = question_datamodel.id
question_text = question_datamodel.text
question = Question (
id=QuestionId(question_id),
question_text=QuestionText(question_text)
)
return question
def transferToDataModel(self, question:Question):
question_id = question.id.get_id()
question_text = question.question_text.get_text()
question_datamodel = QuestionDataModel(
id=question_id,
text=question_text
)
return question_datamodel
toModelとtransferToDataModelではデータの詰め替えを行っています。
FastAPI としては DataModel でないとフレームワークの仕様に沿わないので都合が悪いです。
一方で、開発者としては DDD で解釈したい気持ちがあります。
ですので、その変換を Repository のレイヤーで行い必要に応じてデータの詰め替えをします。
イメージとしては以下のような感じです。
ポイント3: テストのための InMemoryRepository の実装
確認その 1) DDD における Repository の役割
DDDにおけるリポジトリの役割はデータの永続化と再構築です。
SQL 的に言えば、データの永続化 insert, update, delete, 再構築は select になります。
確認その 2) DDD における Repository のメリット
今まで自分の開発の経験として、DBMS(MySQL など)を利用したテストは書いていました。
DBMS を利用するテストにはデメリットがあります。
それは DBMS を利用して直接、データを再構築するようなテストは、実行するサーバーや実行するタイミングといった環境の違いによって結果がばらつくことです。環境による不都合が生じてしまいます。
DDDもしくはレイヤーに分けることで、それぞれのレイヤーで開発するフレームワークに依存しないテストができることがメリットの一つだと思います。
以下では、実際にコードを示していきます。
ORM をラップしたリポジトリの実装
今回、FastAPI では ORM として SQLalchemy を利用しています。
以下のコードでは Question を保存(永続化)する処理を実装しています。
テストを書くときにこのリポジトリを利用すると、DB に対して実際に登録することになり前述したような環境による不都合が起きることになります。
from ...value_object.question.question import Question
from ...value_object.question.question_id import QuestionId
from ...value_object.question.question_text import QuestionText
from ...value_object.application.application_id import ApplicationId
from ...value_object.application.application_code import ApplicationCode
from .i_question_repository import IQuestionRepository
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import func
from sqlalchemy.dialects.postgresql import insert
import pdb
import sys
import os
sys.path.append(os.path.join(os.path.dirname(__file__), '../../../'))
from database.setting.setting import ENGINE, Base
from database.model.question import QuestionDataModel
class SqlAlchemyQuestionRepository(IQuestionRepository):
(略)
def __init__(self, ENGINE):
self.session = sessionmaker(ENGINE)
def save(self, question:Question):
session = self.session()
insert_statement = insert(QuestionDataModel).values(
id=question.id.get_id(),
text=question.question_text.get_text()
)
upsert_statement = insert_statement.on_conflict_do_update(
index_elements=['id'],
set_=dict(
text=question.question_text.get_text()
)
)
session.execute(upsert_statement)
session.commit()
InMemory なリポジトリの実装
そこで実際に DB にアクセスをしない代わりにメモリ上にデータを保存するリポジトリを実装します。
from ...value_object.question.question import Question
from ...value_object.question.question_id import QuestionId
from .i_question_repository import IQuestionRepository
class InMemoryQuestionRepository(IQuestionRepository):
def __init__(self):
self.store = []
def save(self, question:Question):
self.store += [question]
リポジトリが交換可能であることを担保する
ここで InMemory のリポジトリと DB にアクセスするリポジトリが同じ処理をするもの、交換可能であることを担保するために、以下のような抽象クラスを用意しておきます。（いわゆる、依存性逆転の原則です。）
import abc
from ...value_object.question.question import Question
from ...value_object.question.question_id import QuestionId
class IQuestionRepository(metaclass=abc.ABCMeta):
@abc.abstractmethod
def save(self, question:Question) -> None:
return
このように InMemory のリポジトリと DB にアクセスするリポジトリをそれぞれ用意することで、テストを実行する際に環境による不都合も起きることもなくなり、開発者の精神衛生的にも良いでしょう。
実行速度としても InMemory のリポジトリのほうが速いのもメリットです。
ポイント4: FastAPI から自動生成される swagger と DDD
FastAPI には swagger を自動生成してくれる機能があります。
これが結構便利なのですけど、DDDにそった形でdomain objectを実装してもそれをよしなに認識してくれることはありません。
swagger をそれなりの形にしたい場合は、FastAPI の仕様に則った書き方をする必要があります。
具体的にはこんな感じ。
from fastapi import APIRouter
from typing import List
from ddd.domain.repository.question.sqlalchemy_question_repository import SqlAlchemyQuestionRepository
from ddd.domain.repository.application.sqlalchemy_application_repository import SqlAlchemyApplicationRepository
from ddd.domain.application_service.question.question_application_service import QuestionApplicationService
from ddd.domain.application_service.question.question_search_command import QuestionSearchCommand
from ddd.domain.application_service.question.question_update_command import QuestionUpdateCommand
from database.setting.setting import ENGINE, Base
router = APIRouter()
from pydantic import BaseModel
class CreateQuestion(BaseModel):
text: str
application_code: str
class ResponseQuestion(BaseModel):
id: int
text: str
application_code: str
@router.post("/question", response_model=List[ResponseQuestion])
async def register_question(question: CreateQuestion):
question_repository = SqlAlchemyQuestionRepository(ENGINE)
question_application_service = QuestionApplicationService(question_repository)
question_update_command = QuestionUpdateCommand(
id=None,
question_text=question.text
)
questions = question_application_service.register_question(question_update_command)
question_datamodels = list(map(lambda question: question_repository.transferToDataModel(question), questions))
question_responses = list(map(lambda question: ResponseQuestion(id=question.id, text=question.text, application_code=question.application_code), question_datamodels))
return question_responses
swagger を確認する
リクエスト、レスポンスをそれぞれクラスとして定義し、メソッドのアノテーション、引数につけると swagger はいい感じに出力されます。
class CreateQuestion(BaseModel):
text: str
application_code: str
class ResponseQuestion(BaseModel):
id: int
text: str
application_code: str
~~~
@router.post("/question", response_model=List[ResponseQuestion])
async def register_question(question: CreateQuestion):
出力される swagger はこんな感じ。もちろん swagger 上から API をコールすることもできます。
ポイント5: pycache のディレクトリが散在する（仕方ない？）
**__pycache__**というディレクトリが散在することになりました。
.gitignoreにその都度書いて、バージョニングの対象外にするのですけど、
これにはなにかいい方法がありそうだと思っています。
知っている方いたら教えてください。
まとめ
DDDの設計思想は FastAPI に限らず、いろんなフレームワークに適用できる部分がとても有用に感じています。
個人的には各レイヤーでテストがかけることにとても助かっています。そんな感じで今回の記事はここまでです。良い開発生活を。
Discussion
内容は同じものですけど、↑こっちにも寄稿しています。