【システム開発戦略編その1】ドメインドリブンでクリーンなアーキテクチャを意識した開発

ネットワークサービス部のid:uwenoです。

今回から新企画「システム開発戦略編」と題しまして、システム開発において私たちが意識している、より抽象度の高い概念・考え方についてご紹介していきます。

第1回目の今回は、ソフトウェアの設計手法であるドメイン駆動設計、およびクリーンアーキテクチャの考え方について簡単に紹介し、私たちのプロダクトでそれらをどのように実装しているかをPythonのサンプルコードを用いて解説します。ドメイン駆動開発やクリーンアーキテクチャについては、私のようなものが語るまでもなく、数多くの著名な書籍やサイト等で紹介されています。そのためこの記事では、前半はドメイン駆動開発やクリーンアーキテクチャごく簡単な説明にとどめ、後半のサンプルコードを用いた実践例の解説により注力しています。また、ドメイン駆動開発やクリーンアーキテクチャで一般的に利用されている用語については、特段注釈なく利用していますのでご承知おきください。

ドメイン駆動開発

ドメイン駆動開発は、ある特定の事業・業務の知識（=ドメイン）を中心としたソフトウェア設計手法です。これはEric Evans氏の著書 Domain-Driven Design (2003) で示された設計手法であり、昨今のモダンなソフトウェア開発では用いられることの多い手法です。日本語訳版『エリック・エヴァンスのドメイン駆動設計』(2013, 翔泳社)も出版されています。例えば、私たちのプロダクトにとってのドメイン、つまりプログラム化の対象となる範囲は、「プライベートブリッジというクラウドサービスにおけるネットワーク運用業務」となります。ドメイン駆動開発の考え方に従うことで、インフラ設定などのドメインと関係ない部分のコードをドメイン層から切り離すことができ、開発者はドメインの問題に対してより集中して取り組むことができるようになります。

クリーンアーキテクチャ

クリーンアーキテクチャは、システムをいくつかの層に分けることで、特定のフレームワークやライブラリの依存を少なくしつつ、変更や拡張をしやすくするためのソフトウェア開発手法です。これはソフトウェア開発界隈では「ボブおじさん」として知られるRobert C. Martin氏のブログ記事(2011)で示されました。各レイヤー間では一方向にしか依存できないため、しばしば複数の円からなる図で示されることがあります。私たちのプロダクトでは、「ドメイン層」「ユースケース層」「インタフェース層」「インフラストラクチャー層」の4つの層を定義し、ソフトウェア開発を実践しています。

Pythonでの実践例

私たちのチームで実践しているPythonでの設計・実装手法を、簡単なToDo管理Webアプリを例に紹介します。実装には「SQLAlchemy」「FastAPI」などの複数のライブラリを利用します。

なお説明に必要な最低限のコードしか記載しておらず、ToDo管理アプリとしての十分な機能（CRUDのすべて）は実装されていません。また、コード中で利用するライブラリの詳細な利用方法は説明しません。

サンプルコードのディレクトリ構成は以下の通りです。

.
├── main.py
├── domain
│   ├── __init__.py
│   ├── model.py
│   └── service.py
├── usecase
│   ├── __init__.py
│   └── dto.py
│   └── usecase.py
├── interface
│   ├── __init__.py
│   ├── repository.py
│   └── router.py
├── infrastructure
│   ├── __init__.py
│   └── db.py
└── di
    ├── __init__.py
    └── injector.py

トップレベルのファイルにはこのプロジェクトのエントリーポイントである main.py があります。また、先ほど紹介した「ドメイン層」「ユースケース層」「インタフェース層」「インフラストラクチャー層」に対応するディレクトリと、外側の層から内側の層に依存性を注入するためのディレクトリから構成されています。

これらのディレクトリの階層、ファイルについて、1ずつ順番に紹介していきます。コードの紹介が長くなってしまい恐縮ですが、最後までお付き合いいただければ幸いです。

ドメイン層

ドメイン層には主に、ドメイン駆動設計でいうところのドメインモデルやドメインサービスを記述します。ドメインの知識はなるべくドメイン層で表現し、逆にドメインとは関係のない情報はドメイン層から除外しています。

下記はドメインモデル(domain/model.py)の実装例です。

from abc import ABC, abstractmethod
from datetime import date
from enum import Enum, auto
from uuid import UUID


class State(Enum):
    open = auto()
    doing = auto()
    close = auto()


class Todo:
    def __init__(self, id_: UUID, title: str, due_date: date, state: State) -> None:
        self._title = title
        self._due_date = due_date
        self._state = state
        self._id = id_

    @property
    def id(self) -> UUID:
        return self._id

    @property
    def title(self) -> str:
        return self._title

    @title.setter
    def title(self, title: str) -> None:
        self._title = title

    @property
    def due_date(self) -> date:
        return self._due_date

    @due_date.setter
    def due_date(self, due_date: date) -> None:
        self._due_date = due_date

    @property
    def state(self) -> State:
        return self._state

    @state.setter
    def state(self, state: State) -> None:
        self._state = state

    @property
    def expired(self) -> bool:
        return date.today() > self.due_date


class TodoRepository(ABC):
    @abstractmethod
    def add(self, todo: Todo) -> None:
        ...

    @abstractmethod
    def get_all(self) -> list[Todo]:
        ...

値オブジェクトとエンティティ
- 今回の例ではState(Enum)が値オブジェクトに該当します
- ここではPythonのクラス（Enum含む）を用いて表現しています
集約ルート（以下Todo集約）
- 今回の例ではTodoの集約ルート（以下Todo集約）が該当します
- 集約はクラスで実装します
- 集約で保持するエンティティや値オブジェクトはプライベートなメンバーとし、境界の外部から直接参照や変更がされないよう、 propertyデコレータでsetterとgetterを実装しています
- 集約のidフィールドにはUUIDなどの一意識別子などを利用し、集約クラスのオブジェクト（エンティティ）の同一性を確保しています
- 単一のオブジェクト（エンティティ）で実装可能なビジネスロジックは、集約クラスのメソッドとして実装します
集約のリポジトリの定義
- 今回の例ではTodo集約のリポジトリの定義が該当します
- 集約を操作するリポジトリは抽象基底クラスとして定義します
  - リポジトリを使って集約をどう扱うか決めるのはドメイン層の役割ですが、リポジトリが具体的にどうやって実装されているかはドメインと関係ありません
  - そのためドメイン層ではシグネチャーのみを定義し、実装はインタフェース層で行います

下記はドメインサービス(domain/service.py)の実装例です。

from domain.model import Todo, TodoRepository


class TodoDomainService:
    def __init__(self, todo_repository: TodoRepository) -> None:
        self._todo_repository = todo_repository

    def get_expired_todo_list(self) -> list[Todo]:
        all_todo = self._todo_repository.get_all()

        return [todo for todo in all_todo if todo.expired]

init時には、必要に応じて集約のリポジトリなどを受け取ります
単一のオブジェクト（エンティティ）では実装できず、複数のオブジェクトを扱う必要があるビジネスロジックは、ドメインサービスクラスのメソッドとして実装します
今回の例ではTodo集約のみを扱っていますが、実際には複数の集約が相互作用するロジックを記述することが多いです

ユースケース層

ユースケース層にはユースケースの実装や、ユースケース層で利用するData transfer object (DTO) の定義などを行います。

下記はDTO(usecase/dto.py)の実装例です。

from datetime import date
from enum import Enum

from pydantic import BaseModel


class StateChoice(str, Enum):
    open = "open"
    doing = "doing"
    close = "close"


class TodoOutput(BaseModel):
    id: str
    title: str
    due_data: date
    state: StateChoice
    expired: bool


class TodoInput(BaseModel):
    title: str
    due_data: date

DTOはpydanticのBaseModelを使って定義しています
このDTOは後述するFastAPIのレスポンスモデルとしても利用します

下記はユースケース(usecase/usecase.py)の実装例です。

import uuid
from datetime import date

from injector import inject

from domain.model import State, Todo, TodoRepository
from domain.service import TodoDomainService
from usecase.dto import StateChoice, TodoInput, TodoOutput


class TodoUsecase:
    @inject
    def __init__(self, todo_repository: TodoRepository) -> None:
        self._todo_repository = todo_repository
        self._todo_domain_service = TodoDomainService(todo_repository=todo_repository)

    def create_todo(self, input: TodoInput) -> TodoOutput:
        todo = Todo(
            id_=uuid.uuid4(),
            title=input.title,
            due_date=input.due_data,
            state=State.open,
        )

        self._todo_repository.add(todo)

        return TodoOutput(
            id=str(todo.id),
            title=todo.title,
            due_data=todo.due_date,
            state=StateChoice[todo.state.name],
            expired=todo.expired,
        )

    def get_expired_todo_list(
        self,
    ) -> list[TodoOutput]:
        return [
            TodoOutput(
                id=str(todo.id),
                title=todo.title,
                due_data=todo.due_date,
                state=StateChoice[todo.state.name],
                expired=todo.expired,
            )
            for todo in self._todo_domain_service.get_expired_todo_list()
        ]

ユースケースはクラスとして実装しています
- initで集約のリポジトリや（今回の例ではありませんが）集約のファクトリーなどを受け取ります
- 同じくinitでドメインサービスのインスタンス化を行います
メソッドとして各ユースケースの処理を記述します
- ユースケースにはビジネスロジックは含めずに、リポジトリやドメインサービスの操作を中心に行います
- 今回の例ではTodoを直接インスタンス化していますが、ファクトリーを呼び出すことも多いです
ユースケースのinputとoutputにDTOを用いることで、ドメイン層のオブジェクトがインタフェース層へ漏れ出さないようにしています

インタフェース層

インタフェース層にはリポジトリやアダプターの実装、Webルーターの定義・実装、DBのスキーマ定義などが含まれます。

下記はリポジトリ(interface/repository.py)の実装例です。

from injector import inject
from sqlalchemy import Column, Date, Engine, Integer, String, insert
from sqlalchemy.orm import declarative_base

from domain.model import Todo, TodoRepository

Base = declarative_base()


class TodoTable(Base):
    __tablename__ = "todo"

    id = Column("id", String(36), primary_key=True)
    title = Column("title", String(256))
    due_date = Column("due_date", Date)
    state = Column("state", Integer)


class TodoRepositoryImpl(TodoRepository):
    @inject
    def __init__(self, engine: Engine) -> None:
        self._engine = engine

    def add(self, todo: Todo) -> None:
        with self._engine.connect() as connection:
            stmt = insert(TodoTable).values(
                id=str(todo.id),
                title=todo.title,
                due_date=todo.due_date,
                state=todo.state.value,
            )
            connection.execute(stmt)
            connection.commit()

    def get_all(self) -> list[Todo]:
        raise NotImplementedError

ドメイン層で定義したTodo集約のリポジトリを実装しています
今回の例ではSQLAlchemyを用いてデータベースにエンティティを保存しています

以下はWebのルーター(interface/router.py)の実装例です。

from abc import ABC, abstractmethod
from datetime import date
from typing import Any

from fastapi import APIRouter, Body
from injector import inject

from usecase.dto import TodoInput, TodoOutput
from usecase.usecase import TodoUsecase


class WebRouter(ABC):
    @abstractmethod
    def get_router(self) -> APIRouter:
        ...


class TodoWebRouter(WebRouter):
    @inject
    def __init__(self, todo_usecase: TodoUsecase) -> None:
        self._todo_usecase = todo_usecase
        self._router = APIRouter()

    async def _create(self, input: TodoInput) -> TodoOutput:
        todo = self._todo_usecase.create_todo(input)
        return todo

    def get_router(self) -> APIRouter:
        self._router.post("/", response_model=TodoOutput)(self._create)

        return self._router

今回の例ではFastAPIを使ってWeb APIのインタフェースを定義しています
WebRouterの定義と実装を分けているのは、後述するDIを書きやすくするためです

インフラストラクチャー層

インフラ層にはインフラにまつわる設定を書きます。下記はデータベース(infrastructure/db.py)の実装例です。

from sqlalchemy import create_engine

from interface.repository import Base

engine = create_engine("sqlite:///:memory:", echo=True)


def create_table() -> None:
    Base.metadata.create_all(engine)

SQlite3のインメモリーデータベースを利用しています
アプリケーション実行時にテーブルを作成するための関数を書いています

依存性の注入

紹介した例の中にもあったように、ユースケース層でドメイン層の集約のリポジトリを利用するケースは頻出します。ドメイン層にはリポジトリの定義しか存在せず、実際にリポジトリを利用するために必要な実装はインタフェース層にあります。しかし、クリーンアーキテクチャに従うと、内側の層から外側の層への依存は許されません。そのため、例えば usecase/usecase.py のファイル中で interface/repository.py に定義された TodoRepositoryImpl をインポートできません。

この問題を解決するために「依存性の注入(Dependency Injection)」を利用します。今回の例ではinjectorという依存性注入フレームワークを活用しています。

下記は依存性を注入する実装例です(di/injector.py)。

from injector import Binder, Injector, Module
from sqlalchemy import Engine

from domain.model import TodoRepository
from infrastructure.db import engine
from interface.repository import TodoRepositoryImpl
from interface.router import TodoWebRouter, WebRouter


class InjectorModule(Module):
    def configure(self, binder: Binder) -> None:
        binder.bind(WebRouter, TodoWebRouter)
        binder.bind(TodoRepository, TodoRepositoryImpl)
        binder.bind(Engine, engine)


def get_injector() -> Injector:
    return Injector(InjectorModule)

注入されるもの	注入するもの
抽象基底クラス `WebRouter`	WebRouterの実装 `TodoWebRouter`
抽象基底クラス `TodoRepository`	TodoRepositoryの実装 `TodoRepositoryImpl`
クラス `Engine`	Engineクラスのオブジェクト `engine`

このインジェクターから TodoRepository を取り出すと、TodoRepository に対してバインドされた TodoRepositoryImpl のインスタンスを取り出すことができます。インスタンス化をする際には、initに必要な引数を、型情報をもとにInjectorから再帰的に取り出してくれます。つまり、TodoRepositoryImpl のインスタンス化時には、Injectorの Engine にバインドされた engine をinitの引数に与えます。これにより、インフラストラクチャー層に実装されたengineを、インタフェース層で利用できます。なお、この機能を利用するためには取り出すクラスのinitメソッドが @inject デコレータで修飾されている必要があります。

おまけ

main.pyとpyproject.tomlの紹介です。

下記は main.py の実装例です。

from fastapi import FastAPI

from di.injector import get_injector
from infrastructure.db import create_table
from interface.router import WebRouter

create_table()

app = FastAPI()

router = get_injector().get(WebRouter).get_router()

app.include_router(router=router)

下記は今回利用したpyproject.tomlの例です。

[tool.poetry]
name = "test-app"
version = "0.1.0"
description = "Architecture sample."
authors = ["uewno"]

[tool.poetry.dependencies]
python = "^3.11"
sqlalchemy = "^2.0.17"
uvicorn = {extras = ["standard"], version = "^0.22.0"}
injector = "^0.20.1"
fastapi = "^0.100.0"
pydantic = "^2.0.2"


[tool.poetry.group.dev.dependencies]
black = "^23.7.0"
isort = "^5.12.0"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

ここまで紹介してきたコードやファイル群を利用し、下記コマンドを実行することで、実際にWeb APIを動作させることができます。

poetry install
poetry run uvicorn main:app

終わりに

今回は私たちがドメイン駆動設計、クリーンアーキテクチャの考え方を、 Pythonのソフトウェアで具体的にどう実践しているかをご紹介しました。

Pythonでドメイン駆動設計を実践したい、クリーンアーキテクチャな構造にチャレンジしたいというかたの参考になれば幸いです。次回はスクラムについて紹介します。ぜひお楽しみに。

FJCT Tech blog

富士通クラウドテクノロジーズ公式エンジニアブログです