- はじめに
- 実行環境
- ディレクトリ構造
- ソースコードと簡単な解説
- 感想
- 最後に
はじめに
テクノロジー本部 デジタルテクノロジー統括部 デジタルソリューション部でエンジニアをしている@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 <sample@sample.com>"] 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月現在の情報です。