🦕

FastAPI(Python)+DDD(ドメイン駆動開発)のポイント!!

2022/05/26に公開

カラビナテクノロジーでサーバーサイドエンジニアをしているブラボーです。
今回はDDD(ドメイン駆動開発)をFastAPIで実装するときのプラクティスを紹介します。


この記事で分かること

  • この記事で分かること

    • FastAPI+DDDのプラクティス(具体的には↓)
      • FastAPIにおけるDDDを取り入れたディレクトリ構成
      • FastAPIのModelとDDDのValue Objectのデータ扱い方(データ変換)
      • テストのためのInMemoryRepositoryの実装
      • FastAPIから自動生成されるswaggerとDDD
      • おまけ)pycacheのディレクトリが散在する
  • この記事で分からないこと

    • FastAPIのゼロからの開発手順
    • DDDのゼロからの知識を学ぶ

前提

言語、ミドルウェアのバージョン

言語、ミドルウェア バージョン
python 3.7.13
FastAPI 0.75.1
sqlalchemy 1.4.35

https://fastapi.tiangolo.com/

https://www.sqlalchemy.org/

例題: アンケートできるサービス

ここで扱う題材として、QAが投稿できるサービスを想定しています。
特にこの記事ではQuestion(質問)する部分についてフォーカスして書いていきます。

オニオンアーキテクチャ

コードの書き方としては、オニオンアーキテクチャをなんとなく参考にしている感じです。

https://dzone.com/articles/onion-architecture-is-interesting

DDDの参考書籍

今回のDDDの実践については、この書籍を参考にしています。

https://www.amazon.co.jp/dp/B082WXZVPC/ref=dp-kindle-redirect?_encoding=UTF8&btkr=1

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_servicerepositoryservicevalue_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_servicerepositoryservicevalue_objectといった各レイヤーのディレクトリを配置しました。
他にも適切なディレクトリ構造が考えつきますが、DDD的な情報の整理はできるので今回はOKとしました。

テストコードの例

以下にそれぞれのレイヤーの実装したコードとテストコードを示します。

value_objectのテストの例
/src/tests/domain/value_object/test_question.py
    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のテストの例
/src/tests/domain/repository/question/test_question_sqlalchemy_repository.py
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 を使うと便利です。

/src/tests/domain/application_service/test_question_application_service.py
    @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
src/ddd/domain/repository/question/sqlalchemy_question_repository.py
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

toModeltransferToDataModelではデータの詰め替えを行っています。
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に対して実際に登録することになり前述したような環境による不都合が起きることになります。

/src/ddd/domain/repository/question/sqlalchemy_question_repository.py
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にアクセスをしないメモリ上にデータを保存するリポジトリを実装します。

/src/ddd/domain/repository/question/in_memory_question_repository.py
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にアクセスするリポジトリが同じ処理をするもの、交換可能であることを担保するために、以下のような抽象クラスを用意しておきます。(いわゆる、依存性逆転の法則です。)

/src/ddd/domain/repository/question/i_question_repository.py
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の仕様に則った書き方をする必要があります。
具体的にはこんな感じ。

/src/routers/question.py
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