FastAPI(Python)+DDD(ドメイン駆動開発)のポイント!!
カラビナテクノロジーでサーバーサイドエンジニアをしているブラボーです。
今回は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の実践については、この書籍を参考にしています。
FastAPIにおけるDDDを取り入れたディレクトリ構成
FastAPI/
|- src/
|- database
|- ddd
|- domain
|- application_service
|- repository
|- service
|- value_object
|- migration
|- routers
|- tests
ディレクトリ(srcを基準) | 説明 |
---|---|
database | FastAPIとしてのModelが格納されます。DDDでいうとDataModelに相当します。 |
ddd/domain | application_service、repository、service、value_objectといった各レイヤーのディレクトリを配置します |
migration |
migrationにはマイグレーションのパッケージであるalembicを利用しました。 https://alembic.sqlalchemy.org/en/latest/ |
routers | routersはFastAPIにおける各APIのルートを担当します。 |
tests |
testsにはdomain、つまりDDD的な意味での各レイヤーのテストが格納されます。 今回はこのプロジェクトではテストツールとしてpytestを利用しました。 https://docs.pytest.org/en/7.1.x/ |
補足
domainディレクトリについて
domainにはapplication_service、repository、service、value_objectといった各レイヤーのディレクトリを配置しました。
他にも適切なディレクトリ構造が考えつきますが、DDD的な情報の整理はできるので今回はOKとしました。
テストコードの例
以下にそれぞれのレイヤーの実装したコードとテストコードを示します。
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()
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のレイヤーで行い必要に応じてデータの詰め替えをします。
イメージとしては以下のような感じです。
テストのためのInMemoryRepositoryの実装
確認その1) DDDにおけるRepositoryの役割
DDDにおけるリポジトリの役割はデータの永続化と再構築です。
SQL的に言えば、データの永続化insert, update, delete, 再構築はselectになります。
確認その2) DDDにおけるRepositoryのメリット
今まで自分の開発の経験として、DBMS(MySQLなど)を利用したテストは書いていました。
DBMSを利用するテストにはデメリットがあります。
それはDBMSを利用して直接、データを再構築するようなテストは、実行するサーバーや実行するタイミングといった環境の違いによって結果がばらつくことです。環境による不都合が生じてしまいます。
DDDもしくはレイヤーに分けることで、それぞれのレイヤーで開発するフレームワークに依存しないテストを作ることができます。
これが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のリポジトリのほうが速いのもメリットです。
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をコールもできます。
おまけ) pycacheのディレクトリが散在する
**__pycache__**というディレクトリが散在することになりました。
.gitignoreにその都度書いて、バージョニングの対象外にするのですけど、
これにはなにかいい方法がありそうだと思っています。
知っている方いたら教えてください。
まとめ
DDDの設計思想はFastAPIに限らず、いろんなフレームワークに適用できる部分がとても有用に感じています。
個人的には各レイヤーでテストがかけることにとても助かっています。そんな感じで今回の記事はここまでです。良い開発生活を。
Discussion