PythonでDDDやってみた💪

技術 エンジニア Python PostgreSQL デジタルテクノロジー統括部 pickup

はじめに

テクノロジー本部 デジタルテクノロジー統括部 デジタルソリューション部でエンジニアをしている@oyanagiです。

今回はPythonのFastAPIとSQLModelを使ってDDD（ドメイン駆動設計）をやってみました。 「なぜPythonで？」と思うかも知れませんが、私が所属しているチームでは主にPythonが使われているからです😅

DDDをやるにあたり、松岡さん（ブログ/X/BOOTH）の書籍などを参考に、多少アレンジを加えて実装しています。

それではさっそく。

実行環境

  • CPU
    • Intel(R) Core(TM) i7-1185G7 @ 3.00GHz
  • メモリ
    • 32GB
  • OS
    • Windows 10 Home
  • IDE
    • PyCharm 2023.2.2 (Professional Edition)
  • 言語
    • Python 3.11
  • パッケージ管理
    • poetry
  • フレームワーク
    • FastAPI
  • アーキテクチャ
    • オニオンアーキテクチャ
  • DB
    • PostgreSQL 14.5
    • SQLModel(SQLAlchemy) + psycopg2
  • テスト
    • pytest
    • SQLmodel(SQLAlchemy)
  • Linter
    • ruff
  • マイグレーション
    • SQLModel(SQLAlchemy) + Alembic + psycopg2

ディレクトリ構造

最上段は次の通り。

ディレクトリ or ファイル 概要
app アプリケーション本体
migrations マイグレーション関連
tests テストケース関連
.gitignore -
alembic.ini Alembicの設定ファイル
conftest.py pytestの設定ファイル
pyproject.toml poetryの設定ファイル
ruff.toml ruffの設定ファイル

今回はDDDの記事なので、基本的にappディレクトリにフォーカスします。

app

アプリケーション本体。

ディレクトリ or ファイル 概要
core 共通で利用するモジュールなどを配置します
ddd DDD（ドメイン駆動設計）を表現します
　+ appilcation アプリケーション層（ユースケース層）
配下にschema、usecaseディレクトリを用意しているため、親ディレクトリ名を「application」としています。
　+ domain ドメイン層
　+ infra インフラ層
　+ presentation プレゼンテーション層
main.py 起動ファイル

appディレクトリ配下の詳細は次の通りです。

app
├── __init__.py
├── core
│   ├── __init__.py
│   ├── abstract
│   │   ├── __init__.py
│   │   ├── transaction_usecase_base.py
│   │   └── usecase_base.py
│   ├── decorator
│   │   ├── __init__.py
│   │   └── transaction.py
│   ├── exception
│   │   ├── __init__.py
│   │   ├── domain_exception.py
│   │   └── usecase_exception.py
│   ├── interface
│   │   ├── __init__.py
│   │   └── i_entity_sql_model.py
│   ├── middleware
│   │   ├── __init__.py
│   │   └── exception_handling_middleware.py
│   └── mixin
│       ├── __init__.py
│       ├── sql_model_generate_mixin.py
│       └── sql_model_update_mixin.py
├── ddd
│   ├── application
│   │   ├── schema
│   │   │   └── student
│   │   │       ├── __init__.py
│   │   │       ├── student_dto.py
│   │   │       └── student_param.py
│   │   └── usecase
│   │       └── student
│   │           ├── __init__.py
│   │           ├── create_student_usecase.py
│   │           ├── delete_student_usecase.py
│   │           ├── get_student_usecase.py
│   │           └── update_student_usecase.py
│   ├── domain
│   │   └── student
│   │       ├── __init__.py
│   │       ├── i_student_repository.py
│   │       ├── student_entity.py
│   │       └── student_id_value_object.py
│   ├── infra
│   │   ├── database
│   │   │   ├── __init__.py
│   │   │   └── db.py
│   │   ├── repository
│   │   │   ├── __init__.py
│   │   │   └── student_repository.py
│   │   └── router
│   │       ├── __init__.py
│   │       └── router.py
│   └── presentation
│       ├── endpoint
│       │   └── students
│       │       ├── __init__.py
│       │       ├── delete_students.py
│       │       ├── get_students.py
│       │       ├── post_students.py
│       │       ├── put_students.py
│       │       └── router.py
│       └── schema
│           └── students
│               ├── __init__.py
│               ├── students_request.py
│               └── students_response.py
└── main.py

migrations/model

マイグレーションで次のファイルを利用しました。
※アプリケーション本体からもDB接続時に利用しています。

  • student_model.py
from datetime import date, datetime

from sqlmodel import SQLModel, Field, Column, text, SmallInteger, Text

from app.core.mixin import SQLModelUpdateMixin, SQLModelGenerateMixin


class StudentModel(SQLModel, SQLModelUpdateMixin, SQLModelGenerateMixin, table=True):
    __tablename__ = "student_t"

    __table_args__ = {
        "comment": "生徒テーブル",
    }

    id: int | None = Field(default=None, primary_key=True, sa_column_kwargs={"comment": "生徒ID"})
    email: str = Field(nullable=False, max_length=256, sa_column_kwargs={"comment": "email"})
    first_name: str = Field(nullable=False, max_length=32, sa_column_kwargs={"comment": "姓"})
    last_name: str = Field(nullable=False, max_length=32, sa_column_kwargs={"comment": "名"})
    first_kana: str = Field(nullable=False, max_length=64, sa_column_kwargs={"comment": "セイ"})
    last_kana: str = Field(nullable=False, max_length=64, sa_column_kwargs={"comment": "メイ"})
    birth: date | None = Field(default=None, sa_column_kwargs={"comment": "生年月日"})
    gender: int | None = Field(default=None, sa_column=Column(type_=SmallInteger, default=None, comment="性別（1:男性、2:女性、3:未回答）"))
    address: str | None = Field(default=None, sa_column=Column(type_=Text, default=None, comment="住所"))
    tel: str | None = Field(default=None, max_length=16, sa_column_kwargs={"comment": "電話番号"})
    created_at: datetime = Field(nullable=False, sa_column_kwargs={"server_default": text("CURRENT_TIMESTAMP"), "comment": "登録日時"})
    updated_at: datetime = Field(nullable=False, sa_column_kwargs={"server_default": text("CURRENT_TIMESTAMP"), "comment": "更新日時"})
    deleted_at: datetime | None = Field(default=None, sa_column_kwargs={"comment": "削除日時"})

pyproject.toml

次のバージョンで検証しています。

[tool.poetry]
name = "teckteckt-ddd-python"
version = "0.1.0"
description = ""
authors = ["sample <[email protected]>"]
readme = "README.md"
packages = [{include = "teckteckt_ddd_python"}]

[tool.poetry.dependencies]
python = "^3.11"
fastapi = "^0.95.0"
uvicorn = "^0.21.1"
sqlmodel = "^0.0.8"
psycopg2-binary = "^2.9.6"

[tool.poetry.group.lint.dependencies]
ruff = "^0.1.5"

[tool.poetry.group.test.dependencies]
pytest = "^7.4.0"
httpx = "^0.24.1"

[tool.poetry.group.migration.dependencies]
alembic = "^1.10.3"

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

ソースコードと簡単な解説

以下、全ソースコードです。

※__init__.pyの中ではファイルをimportする処理を書いてありますが、ここでは省略します。

app/core

app/core/abstract

それぞれユースケース層で継承して、処理を実装します。

  • usecase_base.py
from abc import ABC, abstractmethod
from typing import Any


class UseCaseBase(ABC):

    @abstractmethod
    def execute(self, *args: Any, **kwargs: Any) -> Any:
        pass
  • transaction_usecase_base.py
from abc import abstractmethod
from typing import Any

from sqlmodel import Session, SQLModel

from app.core.abstract import UseCaseBase


class TransactionUseCaseBase(UseCaseBase):

    def __init__(self, db: Session) -> None:
        self._db: Session = db

    @abstractmethod
    def _transaction(self, *args: Any, **kwargs: Any) -> SQLModel:
        pass

    def db(self) -> Session:
        return self._db

プレゼンテーション層からはexecute()メソッドを呼び出してもらいます。 トランザクション処理は継承先の_transaction()メソッド内で行います。

app/core/decorator

ユースケース層でトランザクション処理を行うデコレータです。

  • transaction.py
from collections.abc import Callable
from typing import Any

from app.core.abstract import TransactionUseCaseBase
from app.core.exception import DomainException, UseCaseException


def transaction(func: Callable[..., Any]) -> Callable[..., Any]:

    def wrapper(*args: Any, **kwargs: Any) -> Any:
        usecase: TransactionUseCaseBase = args[0]
        try:
            result = func(*args, **kwargs)
        except DomainException:
            usecase.db().rollback()
            raise
        except Exception as e:
            usecase.db().rollback()
            raise UseCaseException(description=f"{e}") from e
        else:
            usecase.db().commit()
        return result

    return wrapper

app/core/exception

それぞれドメイン層、ユースケース層で利用するExceptionです。

  • domain_exception.py
import json


class DomainException(Exception):

    def __init__(self, status_code: int, description: str, **kwargs) -> None:
        super().__init__(description)
        self.__status_code: int = status_code
        self.__message: dict = {
            "description": description,
        } | kwargs
        self.__detail: dict = {
            "status_code": self.__status_code,
        } | self.__message

    def __str__(self) -> str:
        return json.dumps(self.__detail)

    def status_code(self) -> int:
        return self.__status_code

    def message(self) -> dict:
        return self.__message
  • usecase_exception.py
import json


class UseCaseException(Exception):

    def __init__(self, description: str, **kwargs) -> None:
        super().__init__(description)
        self.__message = {
            "description": description,
        } | kwargs

    def __str__(self) -> str:
        return json.dumps(self.__message)

    def message(self) -> dict:
        return self.__message

app/core/interface

ドメイン層のエンティティで継承して、処理を実装します。

  • i_entity_sql_model.py
from abc import ABC, abstractmethod

from sqlmodel import SQLModel


class IEntitySQLModel(ABC):

    __config__ = {}

    @classmethod
    @abstractmethod
    def from_model(cls, model: SQLModel) -> SQLModel:
        pass

    @abstractmethod
    def to_primitive_dict(self) -> dict:
        pass

from_model()メソッドはDBから読み込んだモデルをエンティティに変換するためのメソッドで、 to_primitive_dict()メソッドはユースケース層からプレゼンテーション層にデータを返却するときにプリミティブ型だけに変換するメソッドです。

app/core/middleware

FastAPIに追加するエラーハンドリング用のミドルウェアです。

  • exception_handling_middleware.py
from fastapi import Request, Response, status
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint

from app.core.exception import DomainException, UseCaseException


class ExceptionHandlingMiddleware(BaseHTTPMiddleware):

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        try:
            response: Response = await call_next(request)
        except DomainException as e:
            response = JSONResponse(
                status_code=e.status_code(),
                content={"detail": e.message()},
            )
        except UseCaseException as e:
            response = JSONResponse(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                content={"detail": e.message()},
            )
        except Exception as e:
            response = JSONResponse(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                content={
                    "detail": {
                        "description": f"{e}",
                    },
                },
            )
        return response

app/core/mixin

ドメイン層のエンティティで継承して使います。

もともとSQLModel（Pydantic）が持っている処理で頻繁に使いそうだったものをMixinとして用意しました。

  • sql_model_generate_mixin.py
from sqlmodel import SQLModel


class SQLModelGenerateMixin:

    __config__ = {}

    @classmethod
    def generate_by(
        cls: type["SQLModelGenerateMixin"],
        generate_from: SQLModel,
        exclude_unset: bool = True,
        **kwargs,
    ) -> "SQLModelGenerateMixin":
        return cls(**generate_from.dict(
            exclude_unset=exclude_unset,
            **kwargs,
        ))
  • sql_model_update_mixin.py
from sqlmodel import SQLModel


class SQLModelUpdateMixin:

    __config__ = {}

    def update_by(
        self,
        update_data: SQLModel,
        exclude_unset: bool = True,
        **kwargs,
    ) -> None:
        for k, v in update_data.dict(
            exclude_unset=exclude_unset,
            **kwargs,
        ).items():
            setattr(self, k, v)

SQLModelGenerateMixin.generate_by()メソッドは第一引数のオブジェクトからMixinしているクラスを生成するときに使います。 SQLModelUpdateMixin.update_by()メソッドは第一引数のオブジェクトでMixinしているクラスを更新するときに使います。

どちらのメソッドもデフォルト引数をexclude_unset=Trueにすることで、モデルの作成時に明示的に設定されなかったフィールドを返される辞書から除外しています。

app/ddd

app/ddd/application

アプリケーション層（ユースケース層）。

app/ddd/application/schema

アプリケーション層（ユースケース層）ではXxxParamクラスでオブジェクトを受け取って、XxxDtoクラスをプレゼンテーション層に返却しています。

app/ddd/application/schema/studnet
  • student_dto.py
from datetime import date, datetime

from sqlmodel import SQLModel


class __BaseDto(SQLModel):
    id: int
    email: str
    first_name: str
    last_name: str
    first_kana: str
    last_kana: str
    birth: date | None = None
    gender: int | None = None
    address: str | None = None
    tel: str | None = None


class CreateStudentDto(__BaseDto):
    pass


class GetStudentDto(__BaseDto):
    pass


class UpdateStudentDto(__BaseDto):
    pass


class DeleteStudentDto(__BaseDto):
    created_at: datetime
    updated_at: datetime
    deleted_at: datetime
  • student_param.py
from datetime import date

from sqlmodel import SQLModel

from app.core.mixin import SQLModelGenerateMixin


class __BaseParam(SQLModel, SQLModelGenerateMixin):
    birth: date | None = None
    gender: int | None = None
    address: str | None = None
    tel: str | None = None


class StudentIdParam(SQLModel):
    id: int


class CreateStudentParam(__BaseParam):
    email: str
    first_name: str
    last_name: str
    first_kana: str
    last_kana: str


class UpdateStudentParam(__BaseParam):
    email: str | None = None
    first_name: str | None = None
    last_name: str | None = None
    first_kana: str | None = None
    last_kana: str | None = None

app/ddd/application/usecase

app/ddd/application/usecase/student
  • create_student_usecase.py
from sqlmodel import Session

from app.core.abstract import TransactionUseCaseBase
from app.core.decorator import transaction
from app.ddd.application.schema.student import CreateStudentDto, CreateStudentParam
from app.ddd.domain.student import IStudentRepository, StudentEntity
from migrations.model import StudentModel


class CreateStudentUseCase(TransactionUseCaseBase):

    def __init__(self, db: Session, student_repository: IStudentRepository) -> None:
        super().__init__(db)
        self.__student_repository = student_repository

    def execute(self, param: CreateStudentParam) -> CreateStudentDto:
        model: StudentModel = self._transaction(param)

        entity: StudentEntity = self.__student_repository.refresh_to_entity(
            model=model,
        )

        return CreateStudentDto(**entity.to_primitive_dict())

    @transaction
    def _transaction(self, param: CreateStudentParam) -> StudentModel:
        entity: StudentEntity = StudentEntity.generate_by(param)

        return self.__student_repository.insert(entity)
  • delete_student_usecase.py
from sqlmodel import Session

from app.core.abstract import TransactionUseCaseBase
from app.core.decorator import transaction
from app.ddd.application.schema.student import DeleteStudentDto, StudentIdParam
from app.ddd.domain.student import (
    IStudentRepository,
    StudentEntity,
    StudentIdValueObject,
)
from migrations.model import StudentModel


class DeleteStudentUseCase(TransactionUseCaseBase):

    def __init__(self, db: Session, student_repository: IStudentRepository) -> None:
        super().__init__(db)
        self.__student_repository = student_repository

    def execute(self, id_param: StudentIdParam) -> DeleteStudentDto:
        model: StudentModel = self._transaction(id_param)

        entity: StudentEntity = self.__student_repository.refresh_to_entity(
            model=model,
        )

        return DeleteStudentDto(**entity.to_primitive_dict())

    @transaction
    def _transaction(self, id_param: StudentIdParam) -> StudentModel:
        student_id: StudentIdValueObject = StudentIdValueObject.generate_by(id_param)

        return self.__student_repository.delete(student_id)
  • get_student_usecase.py
from app.core.abstract import UseCaseBase
from app.ddd.application.schema.student import GetStudentDto, StudentIdParam
from app.ddd.domain.student import (
    IStudentRepository,
    StudentEntity,
    StudentIdValueObject,
)


class GetStudentUseCase(UseCaseBase):

    def __init__(self, student_repository: IStudentRepository) -> None:
        self.__student_repository = student_repository

    def execute(self, id_param: StudentIdParam) -> GetStudentDto:
        student_id: StudentIdValueObject = StudentIdValueObject.generate_by(id_param)

        entity: StudentEntity = self.__student_repository.find_by_id(student_id)

        return GetStudentDto(**entity.to_primitive_dict())
  • update_student_usecase.py
from sqlmodel import Session

from app.core.abstract import TransactionUseCaseBase
from app.core.decorator import transaction
from app.ddd.application.schema.student import (
    StudentIdParam,
    UpdateStudentDto,
    UpdateStudentParam,
)
from app.ddd.domain.student import (
    IStudentRepository,
    StudentEntity,
    StudentIdValueObject,
)
from migrations.model import StudentModel


class UpdateStudentUseCase(TransactionUseCaseBase):

    def __init__(self, db: Session, student_repository: IStudentRepository) -> None:
        super().__init__(db)
        self.__student_repository = student_repository

    def execute(self, id_param: StudentIdParam, param: UpdateStudentParam) -> UpdateStudentDto:
        model: StudentModel = self._transaction(id_param, param)

        entity: StudentEntity = self.__student_repository.refresh_to_entity(
            model=model,
        )

        return UpdateStudentDto(**entity.to_primitive_dict())

    @transaction
    def _transaction(self, id_param: StudentIdParam, param: UpdateStudentParam) -> StudentModel:
        student_id: StudentIdValueObject = StudentIdValueObject.generate_by(id_param)

        entity: StudentEntity = self.__student_repository.find_by_id(student_id)
        entity.update_by(update_data=param)

        return self.__student_repository.update(entity)

app/ddd/domain

ドメイン層。

app/ddd/domain/student

  • i_student_repository.py
from abc import ABC, abstractmethod

from sqlmodel import Session

from app.ddd.domain.student import StudentEntity, StudentIdValueObject
from migrations.model import StudentModel


class IStudentRepository(ABC):

    @abstractmethod
    def __init__(self, db: Session) -> None:
        pass

    @abstractmethod
    def _fetch_by_id(self, student_id: StudentIdValueObject) -> StudentModel | None:
        pass

    @abstractmethod
    def _apply(self, model: StudentModel) -> StudentModel:
        pass

    @abstractmethod
    def find_by_id(self, student_id: StudentIdValueObject) -> StudentEntity:
        pass

    @abstractmethod
    def insert(self, entity: StudentEntity) -> StudentModel:
        pass

    @abstractmethod
    def update(self, entity: StudentEntity) -> StudentModel:
        pass

    @abstractmethod
    def delete(self, student_id: StudentIdValueObject) -> StudentModel:
        pass

    @abstractmethod
    def refresh_to_entity(self, model: StudentModel) -> StudentEntity:
        pass
  • student_entity.py
from datetime import date, datetime

from sqlmodel import SQLModel

from app.core.interface import IEntitySQLModel
from app.core.mixin import SQLModelGenerateMixin, SQLModelUpdateMixin
from app.ddd.domain.student import StudentIdValueObject
from migrations.model import StudentModel


class StudentEntity(SQLModel, IEntitySQLModel, SQLModelUpdateMixin, SQLModelGenerateMixin):

    id: StudentIdValueObject | None = None
    email: str
    first_name: str
    last_name: str
    first_kana: str
    last_kana: str
    birth: date | None = None
    gender: int | None = None
    address: str | None = None
    tel: str | None = None
    created_at: datetime | None = None
    updated_at: datetime | None = None
    deleted_at: datetime | None = None

    @classmethod
    def from_model(cls, model: StudentModel) -> "StudentEntity":
        return StudentEntity(
            id=StudentIdValueObject(id=model.id),
            **model.dict(exclude={"id"}),
        )

    def to_primitive_dict(self) -> dict:
        user_types = {
            "id": self.id.id,
        }
        primitive_types = self.dict(exclude={"id"})
        return user_types | primitive_types
  • student_id_value_object.py
from sqlmodel import SQLModel

from app.core.mixin import SQLModelGenerateMixin


class StudentIdValueObject(SQLModel, SQLModelGenerateMixin):
    id: int

app/ddd/infra

インフラ層。

app/ddd/infra/database

  • db.py
from sqlmodel import Session, create_engine


def get_db_url() -> str:
    dialect = "postgresql"
    driver = "psycopg2"
    user = "postgres"
    password = "postgres"
    host = "localhost"
    port = "5432"
    database = "teckteckt_ddd_python"
    return f"{dialect}+{driver}://{user}:{password}@{host}:{port}/{database}"


__db_engine = create_engine(
    url=get_db_url(),
    echo=True,
)


def get_db() -> Session:
    with Session(__db_engine) as session:
        yield session

app/ddd/infra/repository

  • student_repository.py
from datetime import datetime

from fastapi import status
from sqlalchemy.sql.operators import is_
from sqlmodel import Session, select

from app.core.exception import DomainException
from app.ddd.domain.student import (
    IStudentRepository,
    StudentEntity,
    StudentIdValueObject,
)
from migrations.model import StudentModel


class StudentRepository(IStudentRepository):

    def __init__(self, db: Session) -> None:
        self.__db: Session = db

    def _fetch_by_id(self, student_id: StudentIdValueObject) -> StudentModel | None:
        statement = select(StudentModel)\
            .where(StudentModel.id == student_id.id)\
            .where(is_(StudentModel.deleted_at, None))
        return self.__db.exec(statement).first()

    def _apply(self, model: StudentModel) -> StudentModel:
        self.__db.add(model)
        return model

    def find_by_id(self, student_id: StudentIdValueObject) -> StudentEntity:
        model: StudentModel = self._fetch_by_id(student_id)
        if model is None:
            raise DomainException(
                status_code=status.HTTP_404_NOT_FOUND,
                description="該当する生徒情報が存在しません。",
            )
        return StudentEntity.from_model(model)

    def insert(self, entity: StudentEntity) -> StudentModel:
        model: StudentModel = StudentModel.generate_by(entity)
        return self._apply(model)

    def update(self, entity: StudentEntity) -> StudentModel:
        model: StudentModel = self._fetch_by_id(entity.id)
        if model is None:
            raise DomainException(
                status_code=status.HTTP_404_NOT_FOUND,
                description="該当する生徒情報が存在しません。",
            )

        model.update_by(entity, exclude={"id", "created_at"})
        model.updated_at = datetime.now()

        return self._apply(model)

    def delete(self, student_id: StudentIdValueObject) -> StudentModel:
        model: StudentModel = self._fetch_by_id(student_id)
        if model is None:
            raise DomainException(
                status_code=status.HTTP_404_NOT_FOUND,
                description="該当する生徒情報が存在しません。",
            )

        model.updated_at = datetime.now()
        model.deleted_at = datetime.now()

        return self._apply(model)

    def refresh_to_entity(self, model: StudentModel) -> StudentEntity:
        self.__db.refresh(model)
        return StudentEntity.from_model(model)

app/ddd/infra/router

  • router.py
from fastapi import APIRouter

from app.ddd.presentation.endpoint import (
    students,
)

main_router = APIRouter()

main_router.include_router(students.router, tags=["/students"])

app/ddd/presentation

プレゼンテーション層。

app/ddd/presentation/endpoint

app/ddd/presentation/endpoint/students
  • delete_students.py
from fastapi import Depends
from sqlmodel import Session

from app.ddd.application.schema.student import (
    DeleteStudentDto,
    StudentIdParam,
)
from app.ddd.application.usecase.student import (
    DeleteStudentUseCase,
)
from app.ddd.infra.database import get_db
from app.ddd.infra.repository import StudentRepository
from app.ddd.presentation.endpoint.students import router
from app.ddd.presentation.schema.students import DeleteStudentsResponse


def __usecase_di(db: Session = Depends(get_db)) -> DeleteStudentUseCase:
    return DeleteStudentUseCase(
        db,
        student_repository=StudentRepository(db),
    )


@router.delete(
    path="/students/{student_id}",
    response_model=DeleteStudentsResponse,
)
def delete_students(
    student_id: int,
    usecase: DeleteStudentUseCase = Depends(__usecase_di),
):
    id_param: StudentIdParam = StudentIdParam(id=student_id)

    dto: DeleteStudentDto = usecase.execute(id_param)

    return DeleteStudentsResponse.generate_by(dto)
  • get_students.py
from fastapi import Depends
from sqlmodel import Session

from app.ddd.application.schema.student import GetStudentDto, StudentIdParam
from app.ddd.application.usecase.student import GetStudentUseCase
from app.ddd.infra.database import get_db
from app.ddd.infra.repository import StudentRepository
from app.ddd.presentation.endpoint.students import router
from app.ddd.presentation.schema.students import GetStudentsResponse


def __usecase_di(db: Session = Depends(get_db)) -> GetStudentUseCase:
    return GetStudentUseCase(
        student_repository=StudentRepository(db),
    )


@router.get(
    path="/students/{student_id}",
    response_model=GetStudentsResponse,
)
def get_students(
    student_id: int,
    usecase: GetStudentUseCase = Depends(__usecase_di),
):
    id_param: StudentIdParam = StudentIdParam(id=student_id)

    dto: GetStudentDto = usecase.execute(id_param)

    return GetStudentsResponse.generate_by(dto)
  • post_students.py
from fastapi import Depends
from sqlmodel import Session

from app.ddd.application.schema.student import (
    CreateStudentDto,
    CreateStudentParam,
)
from app.ddd.application.usecase.student import (
    CreateStudentUseCase,
)
from app.ddd.infra.database import get_db
from app.ddd.infra.repository import StudentRepository
from app.ddd.presentation.endpoint.students import router
from app.ddd.presentation.schema.students import (
    PostStudentsRequest,
    PostStudentsResponse,
)


def __usecase_di(db: Session = Depends(get_db)) -> CreateStudentUseCase:
    return CreateStudentUseCase(
        db=db,
        student_repository=StudentRepository(db),
    )


@router.post(
    path="/students",
    response_model=PostStudentsResponse,
)
def post_users(
    request: PostStudentsRequest,
    usecase: CreateStudentUseCase = Depends(__usecase_di),
):
    param: CreateStudentParam = CreateStudentParam.generate_by(request)

    dto: CreateStudentDto = usecase.execute(param)

    return PostStudentsResponse.generate_by(dto)
  • put_students.py
from fastapi import Depends
from sqlmodel import Session

from app.ddd.application.schema.student import (
    StudentIdParam,
    UpdateStudentDto,
    UpdateStudentParam,
)
from app.ddd.application.usecase.student import (
    UpdateStudentUseCase,
)
from app.ddd.infra.database import get_db
from app.ddd.infra.repository import StudentRepository
from app.ddd.presentation.endpoint.students import router
from app.ddd.presentation.schema.students import (
    PutStudentsRequest,
    PutStudentsResponse,
)


def __usecase_di(db: Session = Depends(get_db)) -> UpdateStudentUseCase:
    return UpdateStudentUseCase(
        db=db,
        student_repository=StudentRepository(db),
    )


@router.put(
    path="/students/{student_id}",
    response_model=PutStudentsResponse,
)
def put_students(
    student_id: int,
    request: PutStudentsRequest,
    usecase: UpdateStudentUseCase = Depends(__usecase_di),
):
    id_param: StudentIdParam = StudentIdParam(id=student_id)
    param: UpdateStudentParam = UpdateStudentParam.generate_by(request)

    dto: UpdateStudentDto = usecase.execute(id_param, param)

    return PutStudentsResponse.generate_by(dto)
  • router.py
from fastapi import APIRouter

router = APIRouter()

app/ddd/presentation/schema

app/ddd/presentation/schema/students
  • students_request.py
from datetime import date

from sqlmodel import Field, SQLModel

_EMAIL_MAX_LENGTH = 256
_FIRST_NAME_MAX_LENGTH = 32
_LAST_NAME_MAX_LENGTH = 32
_FIRST_KANA_MAX_LENGTH = 64
_LAST_KANA_MAX_LENGTH = 64
_TEL_MIN_LENGTH = 10
_TEL_MAX_LENGTH = 16


class __BaseRequest(SQLModel):
    birth: date | None = Field(default=None)
    gender: int | None = Field(default=None)
    address: str | None = Field(default=None)
    tel: str | None = Field(default=None, min_length=_TEL_MIN_LENGTH, max_length=_TEL_MAX_LENGTH)


class PostStudentsRequest(__BaseRequest):
    email: str = Field(max_length=_EMAIL_MAX_LENGTH)
    first_name: str = Field(max_length=_FIRST_NAME_MAX_LENGTH)
    last_name: str = Field(max_length=_LAST_NAME_MAX_LENGTH)
    first_kana: str = Field(max_length=_FIRST_KANA_MAX_LENGTH)
    last_kana: str = Field(max_length=_LAST_KANA_MAX_LENGTH)


class PutStudentsRequest(__BaseRequest):
    email: str | None = Field(default=None, max_length=_EMAIL_MAX_LENGTH)
    first_name: str | None = Field(default=None, max_length=_FIRST_NAME_MAX_LENGTH)
    last_name: str | None = Field(default=None, max_length=_LAST_NAME_MAX_LENGTH)
    first_kana: str | None = Field(default=None, max_length=_FIRST_KANA_MAX_LENGTH)
    last_kana: str | None = Field(default=None, max_length=_LAST_KANA_MAX_LENGTH)
  • students_response.py
from datetime import date

from sqlmodel import Field, SQLModel

from app.core.mixin.sql_model_generate_mixin import SQLModelGenerateMixin

_EMAIL_MAX_LENGTH = 256
_FIRST_NAME_MAX_LENGTH = 32
_LAST_NAME_MAX_LENGTH = 32
_FIRST_KANA_MAX_LENGTH = 64
_LAST_KANA_MAX_LENGTH = 64
_TEL_MIN_LENGTH = 10
_TEL_MAX_LENGTH = 16


class __BaseResponse(SQLModel, SQLModelGenerateMixin):
    id: int = Field()
    email: str = Field(max_length=_EMAIL_MAX_LENGTH)
    first_name: str = Field(max_length=_FIRST_NAME_MAX_LENGTH)
    last_name: str = Field(max_length=_LAST_NAME_MAX_LENGTH)
    first_kana: str = Field(max_length=_FIRST_KANA_MAX_LENGTH)
    last_kana: str = Field(max_length=_LAST_KANA_MAX_LENGTH)
    birth: date | None = Field(default=None)
    gender: int | None = Field(default=None)
    address: str | None = Field(default=None)
    tel: str | None = Field(default=None, min_length=_TEL_MIN_LENGTH, max_length=_TEL_MAX_LENGTH)


class PostStudentsResponse(__BaseResponse):
    pass


class GetStudentsResponse(__BaseResponse):
    pass


class PutStudentsResponse(__BaseResponse):
    pass


class DeleteStudentsResponse(SQLModel, SQLModelGenerateMixin):
    id: int = Field()

main.py

  • main.py
import uvicorn
from fastapi import FastAPI

from app.core.middleware import ExceptionHandlingMiddleware
from app.ddd.infra.router import main_router

app = FastAPI()
app.add_middleware(ExceptionHandlingMiddleware)
app.include_router(main_router)


if __name__ == "__main__":
    uvicorn.run(
        app="app.main:app",
        host="127.0.0.1",
        port=8080,
    )

感想

やってみて感じたのは、レイヤー間を移動するときにデータの入れ替えが発生するので、それが少し手間でした。 実際にサービス開発していくと恩恵を感じるんだと思いますが、まだ分からずです😅

それとせっかくFastAPIを使っているので、実際にはプレゼンテーション層のクラスはいろいろと装飾を加えてAPI仕様書（OpenAPI（Swagger））を整えていたり、 db.pyやmain.pyなどでハードコーディングしている箇所はtomlファイルから読み込んだりしています。

今後はSQLModelだけで非同期処理が書けるようになったらそっちに置き換えたり、 パッケージ管理で「rye」も触ってみたいと考えています💪

最後に

身近にドメインエキスパートが居るわけではないですが、 実際にはDDDのエッセンスを取り入れ検証しました。いわゆる軽量DDDとして。

弊社は個人情報などのデータをたくさん取り扱いますし、外部要因（法改正対応など）における対応がありますので、 そういったドメイン知識をドメイン層に集約することで、ビジネス要件の変更に柔軟に対応し、 かつ、DDDの導入によって、エンジニアにとって保守性の高い実装を実現していきたいです。

※この記事はまだ試行錯誤の過程であり、実際に何かを開発しているわけではありません。適切な方向性を模索中です。今後にご期待ください！

@oyanagi

デジタルテクノロジー統括部 デジタルソリューション部 人事エンジニアグループ 兼 人事IT推進部 HRDXグループ

うさぎ好き。12年に1度しか訪れないうさぎ年がもうすぐ終わろうとしています。今が特別な瞬間だからこそ、1日1日を大切にしています！

※2023年12月現在の情報です。