PythonのFastAPIでDDD（ドメイン駆動設計）構成のバックエンドAPIを開発する方法まとめ

まずは以下のコマンドを実行し、各種ファイルを作成します。

$ mkdir fastapi-domain && cd fastapi-domain
$ mkdir -p deploy/docker/local/py && touch deploy/docker/local/py/Dockerfile
$ mkdir -p src/myapp && touch src/myapp/__init__.py src/myapp/main.py
$ touch .env compose.yml

※ローカル開発環境の構築については、いつものようにDockerを利用するため、試したい方は事前にDocker DesktopなどをインストールしてDockerを使える環境を準備して下さい。

次に作成したファイルをそれぞれ以下のように記述します。

・「deploy/docker/local/py/Dockerfile」

FROM python:3.14.0-slim-trixie

# パッケージ管理用のpoetryをインストール
RUN pip3 install --no-cache-dir poetry

# [開発用] 仮想環境を作成
RUN poetry config virtualenvs.in-project true

WORKDIR /py

EXPOSE 9004

※今回はPythonのバージョン「3.14」を使います。各種ライブラリのパッケージ管理には「poetry」を使います。開発時専用のライブラリを入れるためにpoetryのコンフィグ設定で仮想環境設定「virtualenvs.in-project」を有効化しています。

・「src/myapp/main.py」

from fastapi import FastAPI
from fastapi.responses import PlainTextResponse

app = FastAPI()


@app.get("/", response_class=PlainTextResponse)
def root() -> str:
    return "Hello World !!"

・「.env」

ENV=local
TEST_VALUE=test123

※ローカル環境用の環境変数ファイル

・「compose.yml」

services:
  fastapi:
    container_name: fastapi
    build:
      context: .
      dockerfile: ./deploy/docker/local/py/Dockerfile
    volumes:
      - .:/py
    ports:
      - "9004:9004"
    env_file:
      - .env
    tty: true
    stdin_open: true

次に以下のコマンドを実行し、Dockerコンテナをビルドします。

$ docker compose build --no-cache

次に以下のコマンドを実行し、パッケージ管理用のpoetryの初期化をします。

$ docker compose run --rm fastapi poetry init

コマンド実行後、対話形式で各種設定について聞かれるので、「Package name [py]:」は「fastapi-domain」、「Author [None, n to skip]:」は「n」、「Would you like to define your main dependencies interactively? (yes/no) [yes]」は「no」、「Would you like to define your development dependencies interactively? (yes/no) [yes]」は「no」を入力して実行し、それ以外はそのまま実行して進めます。

完了後、poetryの設定ファイル「pyproject.toml」が作成されます。

次に以下のコマンドを実行し、今回利用する各種パッケージをインストールします。

$ docker compose run --rm fastapi poetry add fastapi uvicorn pydantic
$ docker compose run --rm fastapi poetry add --dev ruff mypy

※FastAPI用の「fastapi」、「uvicorn」、「pydantic」、開発専用ライブラリとして「ruff」（フォーマッター + 静的コード解析）、「mypy」（型チェック）を使います。

次にpoetryの設定ファイル「pyproject.toml」を以下のように修正します。

[project]
name = "fastapi-domain"
version = "0.1.0"
description = ""
authors = [
    {name = "Your Name",email = "you@example.com"}
]
readme = "README.md"
requires-python = ">=3.14"
dependencies = [
    "fastapi (>=0.122.0,<0.123.0)",
    "uvicorn (>=0.38.0,<0.39.0)",
    "pydantic (>=2.12.5,<3.0.0)"
]

# パッケージの読み込み設定
packages = [
    { include = "myapp", from = "src" }
]


[build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"]
build-backend = "poetry.core.masonry.api"

[dependency-groups]
dev = [
    "ruff (>=0.14.7,<0.15.0)",
    "mypy (>=1.19.0,<2.0.0)"
]


# =========================================================
# ruffの設定（フォーマッター + 静的コード解析）
# =========================================================
[tool.ruff]
# Pythonのバージョン指定
target-version = "py314"

# チェック対象外のディレクトリ
exclude = [
    ".venv",
    ".mypy_cache",
    ".ruff_cache",
]

# ==========
# lint設定
# ==========
lint.select = [
    "E", # Pycodestyle エラー
    "F", # Pyflakes
    "B", # bugbear（バグ検出）
    "I", # isort（import整理）
    "UP", # pyupgrade（最新構文）
]

# =========================================================
# mypyの設定（型チェック）
# =========================================================
[tool.mypy]
# Pythonのバージョン指定
python_version = "3.14"

# 厳密モードの有効化
strict = true

# 追加の表示設定
warn_unused_configs = true # 無効な設定や使われていない設定を警告として表示
warn_unreachable = true # 到達不可能コードを警告
show_error_context = true # エラー発生箇所の前後コンテキストを表示
show_column_numbers = true # エラー箇所の列番号を表示

# Pydantic v2の設定
plugins = ["pydantic.mypy"]

# チェック対象外のディレクトリ
exclude = [
    ".venv",
    ".mypy_cache",
    ".ruff_cache",
]

# 外部ライブラリは無視 
ignore_missing_imports = true

※[project]にpackagesの設定を追加（srcディレクトリのmyappパッケージを読み込む）、ruffとmypyの設定（ほぼ最小構成のはず！）を追加します。

次にファイル「deploy/docker/local/py/Dockerfile」、「compose.yml」をそれぞれ以下のように修正します。

・「deploy/docker/local/py/Dockerfile」

FROM python:3.14.0-slim-trixie

# タイムゾーン設定
ENV TZ=Asia/Tokyo

# パッケージ管理用のpoetryをインストール
RUN pip3 install --no-cache-dir poetry

# [開発用] 仮想環境を作成
RUN poetry config virtualenvs.in-project true

WORKDIR /py

# poetry.lockから依存関係をインストール
COPY pyproject.toml poetry.lock .
RUN poetry install --no-root

COPY ./src ./src
EXPOSE 9004

・「compose.yml」

services:
  fastapi:
    container_name: fastapi
    build:
      context: .
      dockerfile: ./deploy/docker/local/py/Dockerfile
    command: poetry run uvicorn myapp.main:app --app-dir src --reload --host 0.0.0.0 --port 9004
    volumes:
      - ./pyproject.toml:/py/pyproject.toml
      - ./poetry.lock:/py/poetry.lock
      - ./src:/py/src
    ports:
      - "9004:9004"
    env_file:
      - .env
    tty: true
    stdin_open: true

※commandでオプション「–app-dir src」を付与し、「myapp」がトップレベルになるようにしています。これによってimport時のファイルパスが短くなります。

次に以下のコマンドを実行し、Dockerコンテナの再ビルドおよび起動します。

$ docker compose down
$ docker compose build --no-cache
$ docker compose up -d

次に以下のコマンドを実行し、ログ出力を確認します。

$ docker compose logs

コマンド実行後、以下のようにFastAPIの起動に関するログ出力がされていればOKです。

次にブラウザで「http://localhost:9004」を開き、以下のように「Hello World !!」が出力されていればOKです。

OpenAPIの仕様書を確認する方法

FastAPIではOpenAPI形式の仕様書が自動で作成されるため、確認したい場合はブラウザで「http://localhost:9004/docs」または「http://localhost:9004/redoc」を開くと確認できます。

・「http://localhost:9004/docs」

・「http://localhost:9004/redoc」

コード修正後に利用するコマンド

今回は開発専用ライブラリとして「ruff」（フォーマッター + 静的コード解析）、「mypy」（型チェック）を使えるようにしたため、コードを修正した際は必要に応じて以下のコマンドを実行し、フォーマット統一および、警告やエラーがでていないことを確認して下さい。

・フォーマット修正

$ docker compose exec fastapi poetry run ruff format .

・importの並び替え（フォーマット修正）

$ docker compose exec fastapi poetry run ruff check . --select I --fix

・静的コード解析によるチェック

$ docker compose exec fastapi poetry run ruff check .

・型チェック

$ docker compose exec fastapi poetry run mypy .

共通設定用の各種ファイル作成

次に以下のコマンドを実行し、共通設定用の各種ファイルを作成します。

$ mkdir -p src/myapp/infrastructure/config
$ touch src/myapp/infrastructure/__init__.py src/myapp/infrastructure/config/settings.py src/myapp/infrastructure/config/__init__.py
$ mkdir -p src/myapp/infrastructure/logger
$ touch src/myapp/infrastructure/logger/__init__.py src/myapp/infrastructure/logger/logger.py
$ mkdir -p src/myapp/application/context
$ touch src/myapp/application/__init__.py src/myapp/application/context/__init__.py src/myapp/application/context/request_context.py
$ mkdir -p src/myapp/presentation/middleware
$ touch src/myapp/presentation/__init__.py src/myapp/presentation/middleware/__init__.py src/myapp/presentation/middleware/request_middleware.py
$ mkdir -p src/myapp/presentation/exception
$ touch src/myapp/presentation/exception/__init__.py src/myapp/presentation/exception/definitions.py src/myapp/presentation/exception/handlers.py
$ mkdir -p src/myapp/infrastructure/database
$ touch src/myapp/infrastructure/database/__init__.py src/myapp/infrastructure/database/database.py
$ mkdir -p src/myapp/presentation/schema/common
$ touch src/myapp/presentation/schema/__init__.py src/myapp/presentation/schema/common/error.py src/myapp/presentation/schema/common/__init__.py

※Pythonの場合は、各ディレクトリをパッケージと認識させるための「__init__.py」も作成します。（最近は省略もできるっぽいですが、テストコードなどで影響が出る可能性があったりして、基本的に作った方がいいようです。）

次に作成したファイルをそれぞれ以下のように記述します。

・「src/myapp/infrastructure/config/settings.py」

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    env: str = "local"
    # デフォル値を付けない場合は必須になるので注意！
    test_value: str

    model_config = SettingsConfigDict(
        env_file=".env",
    )


# サーバー起動時にインスタンス化
# .envがない場合はOSから環境変数を読み込む
settings = Settings()


def get_settings() -> Settings:
    return settings

※環境変数を読み込むためのファイルです。

・「src/myapp/infrastructure/config/__init__.py」

from .settings import Settings, get_settings

__all__ = ["Settings", "get_settings"]

・「src/myapp/infrastructure/logger/logger.py」

import logging
from logging.config import dictConfig

from myapp.application.context import request_id
from myapp.infrastructure.config.settings import get_settings

settings = get_settings()


class EnvFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.env = settings.env
        return True

class RequestIDFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id.get()
        return True


LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "filters": {
        "request_id_filter": {"()": RequestIDFilter},
        "env_filter": {"()": EnvFilter},
    },
    "formatters": {
        "default": {
            "format": (
                "%(asctime)s [%(levelname)s] "
                "ENV=%(env)s - %(name)s "
                "[request_id=%(request_id)s] "
                "%(message)s"
            )
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "filters": ["request_id_filter", "env_filter"],
            "formatter": "default",
        }
    },
    "root": {
        "level": "INFO",
        "handlers": ["console"],
    },
}


def init_logging() -> None:
    dictConfig(LOGGING)


def get_logger(name: str = "fastapi-domain") -> logging.Logger:
    return logging.getLogger(name)

※ロガーの設定変更が必要な場合はこのファイルを修正して下さい。

・「src/myapp/infrastructure/logger/__init__.py」

from .logger import get_logger, init_logging

__all__ = ["init_logging", "get_logger"]

※「__init__.py」を設定することで、他のファイルからimportする時のパスを短くできます。

・「src/myapp/application/context/request_context.py」

from contextvars import ContextVar

# リクエストID（リクエスト単位で一意のID）
request_id: ContextVar[str | None] = ContextVar("request_id", default=None)

※リクエスト単位で一意の値を持たせられるようにするため、コンテキスト変数を作成します。

・「src/myapp/application/context/__init__.py」

from .request_context import request_id

__all__ = ["request_id"]

・「src/myapp/presentation/middleware/request_middleware.py」

import uuid

from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import Response

from myapp.application.context import request_id
from myapp.infrastructure.logger import get_logger


class RequestMiddleware(BaseHTTPMiddleware):
    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        # UUIDの取得
        new_uuid = str(uuid.uuid4())

        # リクエストID用コンテキストにUUIDを設定
        request_id.set(new_uuid)

        # レスポンスヘッダーにX-Request-IDを設定
        response = await call_next(request)
        response.headers["X-Request-ID"] = new_uuid

        # リクエスト開始ログ出力
        logger = get_logger()
        logger.info("start request !!")

        return response

※コンテキストとレスポンスヘッダーにリクエストIDを設定します。

・「src/myapp/presentation/middleware/__init__.py」

from .request_middleware import RequestMiddleware

__all__ = ["RequestMiddleware"]

・「src/myapp/presentation/exception/definitions.py」

######################
# 例外定義
######################

# DBエラー
class DatabaseError(Exception):
    def __init__(self, message: str):
        self.message = message
        super().__init__(message)

※カスタム例外を定義します。

・「src/myapp/presentation/exception/handlers.py」

from typing import cast

from fastapi import Request
from fastapi.exception_handlers import request_validation_exception_handler
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

from myapp.infrastructure.logger import get_logger

######################
# 例外用のハンドラー定義
######################


# RequestValidationError用
async def request_validation_error_handler(
    request: Request, exc: Exception
) -> JSONResponse:
    cast_exc = cast(RequestValidationError, exc)
    logger = get_logger()
    logger.warning(f"バリデーションエラー: {cast_exc.errors()}")
    return await request_validation_exception_handler(request, cast_exc)

# バリデーションエラー
async def valid_error_exception_handler(
    request: Request, exc: Exception
) -> JSONResponse:
    return JSONResponse(
        status_code=422,
        content={"detail": str(exc)},
    )


# DBエラー
async def db_error_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    return JSONResponse(
        status_code=500,
        content={"detail": str(exc)},
    )


# 共通エラー
async def general_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    return JSONResponse(
        status_code=500,
        content={"detail": f"Internal Server Error: {str(exc)}"},
    )

※例外発生時の処理を制御させるためのハンドラー設定です。

・「src/myapp/presentation/exception/__init__.py」

from .definitions import DatabaseError
from .handlers import (
    db_error_exception_handler,
    general_exception_handler,
    request_validation_error_handler,
    valid_error_exception_handler,
)

__all__ = [
    "request_validation_error_handler",
    "valid_error_exception_handler",
    "DatabaseError",
    "db_error_exception_handler",
    "general_exception_handler",
]

・「src/myapp/infrastructure/database/database.py」

# 今回はDBインスタンスはダミーとする。
def get_db() -> str:
    return "DBインスタンスのダミー"

※今回はDBは使わないで進めるため、ダミーとして文字列を返すように定義する。実際にDBを使う場合は、インスタンスを返すようにして下さい。

・「src/myapp/infrastructure/database/__init__.py」

from .database import get_db

__all__ = ["get_db"]

・「src/myapp/presentation/schema/common/error.py」

from pydantic import BaseModel, ConfigDict


# 400 Bad Request
class BadRequestResponse(BaseModel):
    detail: str

    # OpenAPI用設定
    model_config = ConfigDict(
        # 書き換え不可設定
        frozen=True,
        # 「Example Value」に表示されるJSON
        json_schema_extra={
            "example": {
                "detail": "Bad Request",
            }
        },
    )


# 401 Unauthorized
class UnauthorizedResponse(BaseModel):
    detail: str

    # OpenAPI用設定
    model_config = ConfigDict(
        # 書き換え不可設定
        frozen=True,
        # 「Example Value」に表示されるJSON
        json_schema_extra={
            "example": {
                "detail": "Unauthorized",
            }
        },
    )


# 422 Unprocessable Entity
class UnprocessableEntityResponse(BaseModel):
    detail: str

    # OpenAPI用設定
    model_config = ConfigDict(
        # 書き換え不可設定
        frozen=True,
        # 「Example Value」に表示されるJSON
        json_schema_extra={
            "example": {
                "detail": "Unprocessable Entity",
            }
        },
    )


# 500 Internal Server Error
class InternalServerErrorResponse(BaseModel):
    detail: str

    # OpenAPI用設定
    model_config = ConfigDict(
        # 書き換え不可設定
        frozen=True,
        # 「Example Value」に表示されるJSON
        json_schema_extra={
            "example": {
                "detail": "Internal Server Error",
            }
        },
    )

※主にOpenAPI用に共通エラー用のスキーマを定義します。

・「src/myapp/presentation/schema/common/__init__.py」

from .error import (
    BadRequestResponse,
    InternalServerErrorResponse,
    UnauthorizedResponse,
    UnprocessableEntityResponse,
)

__all__ = [
    "BadRequestResponse",
    "UnauthorizedResponse",
    "UnprocessableEntityResponse",
    "InternalServerErrorResponse",
]

DDD（ドメイン駆動設計）のディレクトリ構成について

この後にDDD（ドメイン駆動設計）でAPIを作成していきますが、ディレクトリ構成としてはDDDの思想に基づいたレイヤードアーキテクチャを採用しています。

/src
 ├── /myapp
 |    ├── /application（アプリケーション層）
 |    |    ├── /context（独立した値を持たせるためのコンテキスト変数）
 |    |    └── /usecase（ユースケース層）
 |    |
 |    ├── /di（依存注入によってユースケースのインスタンスをまとめる）
 |    |
 |    ├── /domain（ドメイン層）
 |    |    ├── entity（ドメインモデルの定義）
 |    |    ├── ValueObject（意味のある値を扱うためのオブジェクト定義）
 |    |    ├── repository（リポジトリのインターフェース定義）
 |    |    └── （仮）service（外部サービスのインターフェース定義）
 |    |
 |    ├── /infrastructure（インフラストラクチャー層）
 |    |    ├── /database（データベース設定）
 |    |    ├── /logger（ロガーの実装）
 |    |    ├── /persistence（リポジトリの実装。DB操作による永続化層。）
 |    |    ├── （仮）/cache（キャッシュを含めたリポジトリの実装。インターフェースはリポジトリと同一。）
 |    |    └── （仮）/externalapi（外部サービスの実装）
 |    |
 |    └── /presentation（プレゼンテーション層）
 |         ├── /exception（カスタム例外定義）
 |         ├── /handler（ハンドラー層）
 |         ├── /middleware（ミドルウェア定義）
 |         ├── /router（ルーター定義）
 |         └── /schema（APIの入出力の仕様を決めるスキーマ定義）
 |
 └── /tests（テストコード用ディレクトリ）

※（仮）のものは将来的に追加する想定の例です。

Postドメインを例にAPIを作る

次に以下の手順でPostドメインを例にAPIを作成します。

ドメインの定義

まずは以下のコマンドを実行し、各種ファイルを作成します。

$ mkdir src/myapp/domain/post
$ touch src/myapp/domain/__init__.py src/myapp/domain/post/text.py src/myapp/domain/post/entity.py src/myapp/domain/post/repository.py src/myapp/domain/post/__init__.py

次に作成したファイルをそれぞれ以下のように記述します。

・「src/myapp/domain/post/text.py」

# 型ヒントでまだ定義されていないクラス名を文字列として書けるようにする設定
# これによってクラス同士が互いを参照するような場合でも型ヒントが使いやすくなる
from __future__ import annotations

from dataclasses import dataclass


# 値オブジェクトの定義（フィールドは変更不可設定）
@dataclass(frozen=True)
    class Text:
    value: str

def __post_init__(self) -> None:
    ######################
    # バリデーションチェック
    ######################
    if not self.value:
        raise ValueError("テキストの値は必須です。")

    if len(self.value) > 20:
        raise ValueError("テキストの値は20文字以内で入力して下さい。")

    # オブジェクトを文字列として表示する際に呼ばれるメソッド
    # print() や str() を使う時にこのメソッドの戻り値が表示されます
    def __str__(self) -> str:
        return self.value

    # DBからの復元用クラスメソッド（チェック処理無し）
    @classmethod
    def from_raw(cls, raw_value: str) -> Text:
        # __init__を呼ばずにインスタンス作成した値を設定
        obj = cls.__new__(cls)
        object.__setattr__(obj, "value", raw_value)
        return obj

※今回はPostエンティティ（モデル）のフィールド「text」に意味があることを想定して値オブジェクト（意味のある値を扱うためのオブジェクト）を定義します。新規データ登録時は必ずバリデーションチェックが必要ですが、DB登録後のデータ取得時は基本的にバリデーションチェックは不要なため、復元用のメソッド（from_raw）も定義します。

・「src/myapp/domain/post/entity.py」

# 型ヒントでまだ定義されていないクラス名を文字列として書けるようにする設定
# これによってクラス同士が互いを参照するような場合でも型ヒントが使いやすくなる
from __future__ import annotations

from dataclasses import dataclass

from .text import Text


@dataclass
class Post:
    text: Text
    id: int = 0

    def __post_init__(self) -> None:
    ######################
    # バリデーションチェック
    ######################
    if self.text is None:
        raise ValueError("textは必須です。")

    # DBからの復元用クラスメソッド（チェック処理無し）
    @classmethod
    def from_raw(cls, raw_id: int, raw_text: Text) -> Post:
        # __init__を呼ばずにインスタンス作成した値を設定
        obj = cls.__new__(cls)
        obj.id = raw_id
        obj.text = raw_text
        return obj

※Postエンティティ（モデル）を定義します。フィールド「text」の型には上記で定義した値オブジェクトを利用します。

・「src/myapp/domain/post/repository.py」

from abc import ABC, abstractmethod

from .entity import Post


class PostRepository(ABC):
    # 今回はDB設定はダミー値を使っているため、dbの型はstrにしている
    @abstractmethod
    async def create(self, db: str, post: Post) -> Post:
        pass

    # 今回はDB設定はダミー値を使っているため、dbの型はstrにしている
    @abstractmethod
    async def find_all(self, db: str) -> list[Post]:
        pass

※Postモデルのリポジトリ用の抽象クラス（他のクラスに継承し、定義したメソッドを持つことを強制させる設計図のようなもの）を定義します。今回は例として新規データ作成用と全データ取得用の2種類のみ定義します。ユースケース層でトランザクション管理をするため、メソッドでDBインスタンスを渡せる設計にしてますが、今回はDB設定はダミー値を使うのでstr型で設定してます。

・「src/myapp/domain/post/__init__.py」

from .entity import Post
from .repository import PostRepository
from .text import Text

__all__ = ["Post", "PostRepository", "Text"]

リポジトリの実装

次に以下のコマンドを実行し、リポジトリ実装用のファイルを作成します。

$ mkdir -p src/myapp/infrastructure/persistence/post
$ touch src/myapp/infrastructure/persistence/__init__.py src/myapp/infrastructure/persistence/post/post_repo_impl.py src/myapp/infrastructure/persistence/post/__init__.py

次に作成したファイルをそれぞれ以下のように記述します。

・「src/myapp/infrastructure/persistence/post/post_repo_impl.py」

from myapp.domain.post import Post, PostRepository, Text
from myapp.presentation.exception import DatabaseError


class PostRepositoryImpl(PostRepository):
    # 今回はDB設定はダミー値を使っているため、dbの型はstrにしている
    async def create(self, db str, post: Post) -> Post:
        try:
            # DBへデータ登録を完了した想定でpostを返す
            return post

        except Exception as e:
            raise DatabaseError(f"DBエラーが発生しました。: {str(e)}") from e

    # 今回はDB設定はダミー値を使っているため、dbの型はstrにしている
    async def find_all(self, db str) -> list[Post]:
        try:
            # DBからデータを取得した想定で固定値を返す
            posts = [
                Post.from_raw(
                    raw_id=1, raw_text=Text.from_raw(raw_value="Postデータ１")
                ),
                Post.from_raw(
                    raw_id=2, raw_text=Text.from_raw(raw_value="Postデータ２")
                ),
            ]

            return posts

        except Exception as e:
            raise DatabaseError(f"DBエラーが発生しました。: {str(e)}") from e

※今回はDBを使わないので固定値で返しますが、DBから取得したデータはfrom_rawメソッドを利用してドメインに変換して返します。

・「src/myapp/infrastructure/persistence/post/__init__.py」

from .post_repo_impl import PostRepositoryImpl

__all__ = ["PostRepositoryImpl"]

スキーマの定義

次に以下のコマンドを実行し、APIの入出力の仕様を決めるスキーマ定義用の各種ファイルを作成します。

$ mkdir -p src/myapp/presentation/schema/post
$ touch src/myapp/presentation/schema/post/request.py src/myapp/presentation/schema/post/response.py src/myapp/presentation/schema/post/__init__.py

次に作成したファイルをそれぞれ以下のように記述します。

・「src/myapp/presentation/schema/post/request.py」

from pydantic import BaseModel, ConfigDict, Field


class PostCreateRequest(BaseModel):
    text: str = Field(..., max_length=20, description="テキスト")

    # OpenAPI用設定
    model_config = ConfigDict(
        # 書き換え不可設定
        frozen=True,
        # 「Example Value」に表示されるJSON
        json_schema_extra={
            "example": {
                "text": "Postテキスト",
            }
        },
    )

※リクエスト用のスキーマ定義の場合、各種フィールどには「Field()」を使ってバリデーションチェックの条件を付けます。

・「src/myapp/presentation/schema/post/response.py」

from pydantic import BaseModel


class PostResponse(BaseModel):
    id: int
    text: str

※Postデータをレスポンス結果として返すためのスキーマ定義

・「src/myapp/presentation/schema/post/__init__.py」

from .request import PostCreateRequest
from .response import PostResponse

__all__ = ["PostCreateRequest", "PostResponse"]

ユースケースの定義

次に以下のコマンドを実行し、ユースケース用のファイルを作成します。

$ mkdir -p src/myapp/application/usecase/post
$ touch src/myapp/application/usecase/__init__.py src/myapp/application/usecase/post/create_post.py src/myapp/application/usecase/post/get_posts.py src/myapp/application/usecase/post/__init__.py

次に作成したファイルをそれぞれ以下のように記述します。

・「src/myapp/application/usecase/post/create_post.py」

import logging

from fastapi import HTTPException

from myapp.domain.post import Post, PostRepository, Text
from myapp.presentation.schema.post import PostCreateRequest, PostResponse


class CreatePostUsecase:
    # コンストラクタで依存注入する
    # 今回はDB設定はダミー値を使っているため、dbの型はstrにしている
    def __init__(
        self, logger: logging.Logger, db: str, post_repo: PostRepository
    ) -> None:
        self.logger = logger
        self.db = db
        self.post_repo = post_repo

    async def execute(self, req: PostCreateRequest) -> PostResponse:
        try:
            # 新規Postデータ作成
            newPost = Post(text=Text(req.text))

            # DB登録
            post = await self.post_repo.create(self.db, newPost)
            return PostResponse(id=post.id, text=post.text.value)

        # ValueError発生時の例外処理
        except ValueError as e:
            self.logger.warning("バリデーションエラー: %s", e)
            raise HTTPException(
                status_code=422, detail=f"バリデーションエラー: {str(e)}"
            ) from e

        except Exception as e:
            self.logger.error("DB登録に失敗しました。: %s", e)
            raise e

・「src/myapp/application/usecase/post/get_posts.py」

import logging

from myapp.domain.post import PostRepository
from myapp.presentation.schema.post import PostResponse


class GetPostsUsecase:
    # コンストラクタで依存注入する
    # 今回はDB設定はダミー値を使っているため、dbの型はstrにしている
    def __init__(
        self, logger: logging.Logger, db: str, post_repo: PostRepository
    ) -> None:
        self.logger = logger
        self.db = db
        self.post_repo = post_repo

    async def execute(self) -> list[PostResponse]:
        try:
            # データ取得
            posts = await self.post_repo.find_all(self.db)
            return [PostResponse(id=p.id, text=p.text.value) for p in posts]

        except Exception as e:
            self.logger.error("DBからのデータ取得に失敗しました。: %s", e)
            raise e

・「src/myapp/application/usecase/post/__init__.py」

from .create_post import CreatePostUsecase
from .get_posts import GetPostsUsecase

__all__ = ["CreatePostUsecase", "GetPostsUsecase"]

DIコンテナの作成

次に以下のコマンドを実行し、依存注入によってユースケースのインスタンスをまとめるためのファイルを作成します。

$ mkdir -p src/myapp/di
$ touch src/myapp/di/container.py src/myapp/di/__init__.py

次に作成したファイルを以下のように記述します。

・「src/myapp/di/container.py」

import logging
from typing import Annotated

from fastapi import Depends

from myapp.application.usecase.post import CreatePostUsecase, GetPostsUsecase
from myapp.domain.post import PostRepository
from myapp.infrastructure.database import get_db
from myapp.infrastructure.logger import get_logger
from myapp.infrastructure.persistence.post import PostRepositoryImpl


# リポジトリのDI用関数
def get_post_repository() -> PostRepository:
    return PostRepositoryImpl()


# Post用ユースケース
def create_post_usecase(
    logger: Annotated[logging.Logger, Depends(get_logger)],
    # 今回はDB設定はダミー値を使っているため、dbの型はstrにしている
    db: Annotated[str, Depends(get_db)],
    post_repo: Annotated[PostRepository, Depends(get_post_repository)],
) -> CreatePostUsecase:
    return CreatePostUsecase(logger=logger, db=db, post_repo=post_repo)


def get_posts_usecase(
    logger: Annotated[logging.Logger, Depends(get_logger)],
    # 今回はDB設定はダミー値を使っているため、dbの型はstrにしている
    db: Annotated[str, Depends(get_db)],
    post_repo: Annotated[PostRepository, Depends(get_post_repository)],
) -> GetPostsUsecase:
    return GetPostsUsecase(logger=logger, db=db, post_repo=post_repo)

※FastAPI用のDependsを使って依存注入しています。今回使ったPythonのバージョンでは「Annotated」を使った依存注入が最適でした。（Pythonのバージョンによってここら辺の書き方が変わるので注意）

ハンドラーの定義

次に以下のコマンドを実行し、ハンドラー用のファイルを作成します。

$ mkdir -p src/myapp/presentation/handler/post
$ touch src/myapp/presentation/handler/__init__.py src/myapp/presentation/handler/post/post.py src/myapp/presentation/handler/post/__init__.py

次に作成したファイルをそれぞれ以下のように記述します。

・「src/myapp/presentation/handler/post/post.py」

from typing import Annotated

from fastapi import Depends

from myapp.application.usecase.post import CreatePostUsecase, GetPostsUsecase
from myapp.di.container import create_post_usecase, get_posts_usecase
from myapp.presentation.schema.post import PostCreateRequest, PostResponse


class CreatePostHandler:
    def __init__(
        self,
        usecase: Annotated[CreatePostUsecase, Depends(create_post_usecase)],
    ):
        self.usecase = usecase

    async def execute(self, req: PostCreateRequest) -> PostResponse:
        return await self.usecase.execute(req)


class GetPostsHandler:
    def __init__(
        self,
        usecase: Annotated[GetPostsUsecase, Depends(get_posts_usecase)],
    ):
        self.usecase = usecase

    async def execute(self) -> list[PostResponse]:
        return await self.usecase.execute()

※DIコンテナを使ってハンドラーに依存注入して実行します。

・「src/myapp/presentation/handler/post/__init__.py」

from .post import CreatePostHandler, GetPostsHandler

__all__ = ["CreatePostHandler", "GetPostsHandler"]

ルーター設定

次に以下のコマンドを実行し、ルーター設定用のファイルを作成します。

$ mkdir -p src/myapp/presentation/router
$ touch src/myapp/presentation/router/router.py src/myapp/presentation/router/__init__.py

次に作成したファイルをそれぞれ以下のように記述します。

・「src/myapp/presentation/router/router.py」

from typing import Annotated, Any

from fastapi import APIRouter, Depends

from myapp.presentation.handler.post import CreatePostHandler, GetPostsHandler
from myapp.presentation.schema.common import InternalServerErrorResponse
from myapp.presentation.schema.post import PostCreateRequest, PostResponse

router = APIRouter()

# OpenAPI用の共通エラーレスポンス定義
common_error_res: dict[int | str, dict[str, Any]] | None = {
    500: {"description": "サーバーエラー", "model": InternalServerErrorResponse},
}


@router.post(
    "/post",
    response_model=PostResponse,
    status_code=201,
    responses=common_error_res,
    summary="Postデータ新規作成",
    description="Postデータを新規作成する",
    tags=["Post"],
)
async def create_post(
    req: PostCreateRequest,
    handler: Annotated[CreatePostHandler, Depends(CreatePostHandler)],
) -> PostResponse:
    return await handler.execute(req)


@router.get(
    "/posts",
    response_model=list[PostResponse],
    status_code=200,
    responses=common_error_res,
    summary="Postデータ全件取得",
    description="Postデータを全件取得する",
    tags=["Post"],
)
async def get_posts(
    handler: Annotated[GetPostsHandler, Depends(GetPostsHandler)],
) -> list[PostResponse]:
    return await handler.execute()

※ハンドラーをルーター設定に依存注入して実行します。また@router部分ではOpenAPI用の定義も記述しています。

・「src/myapp/presentation/router/__init__.py」

from .router import router

__all__ = ["router"]

main.pyの修正

次にファイル「src/myapp/main.py」を以下のように修正します。

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware

from myapp.infrastructure.logger import get_logger, init_logging
from myapp.presentation.exception import (
    DatabaseError,
    db_error_exception_handler,
    general_exception_handler,
    request_validation_error_handler,
    valid_error_exception_handler,
)
from myapp.presentation.middleware import RequestMiddleware
from myapp.presentation.router import router


##########################
# アプリ全体の共通初期化処理
##########################
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    ############
    # 起動時処理
    ############
    # ロガーの初期化
    init_logging()

    # サーバー起動ログ出力
    logger = get_logger()
    logger.info("start server !!!")
    # アプリ起動
    yield
    ############
    # 終了時処理
    ############
    # サーバー停止ログ出力
    logger.info("stop server !!!")


#############
# ルーター設定
#############
app = FastAPI(
    lifespan=lifespan,
    title="fastapi-domain API",
    description="FastAPIによるDDD構成のAPIです。",
    version="1.0.0",
    # terms_of_service="https://example.com/terms/",
    # contact={
    # "name": "サポート",
    # "url": "https://example.com/contact/",
    # "email": "support@example.com",
    # },
    # license_info={
    # "name": "Apache 2.0",
    # "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
    # },
)
app.include_router(router)

###################
# ミドルウェアの設定
###################

# 許可したいオリジン
origins = [
    "http://localhost:3000",
]

# CORS設定
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["Content-Type", "Authorization"],
)

# リクエスト用ミドルウェア
app.add_middleware(RequestMiddleware)


###################
# 例外ハンドラー設定
###################
app.add_exception_handler(RequestValidationError, request_validation_error_handler)
app.add_exception_handler(DatabaseError, db_error_exception_handler)
app.add_exception_handler(ValueError, valid_error_exception_handler)
app.add_exception_handler(Exception, general_exception_handler)

※各種共通用設定も記述しています。

次に以下のコマンドを実行し、フォーマット修正、静的コード解析、型チェックを行い、警告が出ないことを確認します。

$ docker compose exec fastapi poetry run ruff format .
$ docker compose exec fastapi poetry run ruff check . --select I --fix
$ docker compose exec fastapi poetry run ruff check .
$ docker compose exec fastapi poetry run mypy .

Dockerコンテナの再ビルドと起動

次に以下のコマンドを実行し、Dockerコンテナを再ビルドします。

$ docker compose down
$ docker compose build --no-cache

次に以下のコマンドを実行し、Dockerコンテナを起動します。

$ docker compose up -d

PostドメインのAPIを試す

次に上記で作成したPostドメインのAPIをPostmanを使って試します。

まずはPOSTメソッドで「http://localhost:9004/post」を実行し、下図のようにステータスコード201で想定通りの結果になればOKです。

次にリクエストパラメータ「text」を21文字以上にして再度実行し、バリデーションチェックでエラーになればOKです。

次にGETメソッドで「http://localhost:9004/posts」を実行し、下図のようにステータスコード200で想定通りの結果になればOKです。

テストコードを追加

次にテストコードを追加して試しますが、まずは以下のコマンドを実行し、テストコードに必要なライブラリをインストールします。

$ docker compose exec fastapi poetry add --dev pytest pytest-asyncio pytest-mock httpx

次にpytestの設定をするため、ファイル「pyproject.toml」の末尾に以下のような設定を追加します。

・「pyproject.toml」

・・・

# =========================================================
# pytestの設定
# =========================================================
[tool.pytest.ini_options]
# パス設定
pythonpath = ["src"]
testpaths = ["src/tests"]

# マーカー設定
markers = [
    "unit: mark a test as a unit test",
    "integration: mark a test as an integration test"
]

次に以下のコマンドを実行し、Dockerコンテナを再ビルドして起動します。

$ docker compose down
$ docker compose build --no-cache
$ docker compose up -d

次に以下のコマンドを実行し、テストコード用のファイルを作成します。

$ mkdir -p src/tests/domain/post
$ touch src/tests/__init__.py src/tests/domain/__init__.py src/tests/domain/post/test_text.py src/tests/domain/post/test_entity.py src/tests/domain/post/__init__.py
$ mkdir -p src/tests/application/usecase/post
$ touch src/tests/application/__init__.py src/tests/application/usecase/__init__.py src/tests/application/usecase/post/test_create_post.py src/tests/application/usecase/post/test_get_posts.py src/tests/application/usecase/post/__init__.py
$ mkdir -p src/tests/presentation/handler/post
$ touch src/tests/presentation/__init__.py src/tests/presentation/handler/__init__.py src/tests/presentation/handler/post/test_post.py src/tests/presentation/handler/post/__init__.py

次に作成したファイルをそれぞれ以下のように記述します。

・「src/tests/domain/post/test_text.py」

import dataclasses

import pytest

from myapp.domain.post import Text


@pytest.mark.unit
class TestText:
    def test_create_text_success(self) -> None:
        text = Text("hello")
        assert text.value == "hello"
        assert str(text) == "hello"

    def test_empty_value_should_raise_error(self) -> None:
        with pytest.raises(ValueError, match="テキストの値は必須です。"):
            Text("")

    def test_over_20_chars_should_raise_error(self) -> None:
        over_text = "a" * 21
        with pytest.raises(
            ValueError, match="テキストの値は20文字以内で入力して下さい。"
        ):
            Text(over_text)

    def test_exactly_20_chars_is_ok(self) -> None:
        t = Text("a" * 20)
        assert t.value == "a" * 20

    def test_frozen_object_cannot_be_modified(self) -> None:
        t = Text("hello")
        with pytest.raises(dataclasses.FrozenInstanceError):
            t.value = "changed" # type: ignore[misc]

    def test_from_raw_should_skip_validation(self) -> None:
        raw = "a" * 100
            t = Text.from_raw(raw)
        assert t.value == raw
        assert isinstance(t, Text)

※テスト実行時にユニットテストのみを指定できるようにするため、「@pytest.mark.unit」を付けています。mypyの型チェックでのエラーをスキップするため、対象箇所に「# type: ignore[misc]」を付けています。

・「src/tests/domain/post/test_entity.py」

import pytest

from myapp.domain.post import Post, Text


@pytest.mark.unit
class TestPost:
    def test_create_post_success(self) -> None:
        text = Text("hello")
        post = Post(text=text, id=1)

        assert post.id == 1
        assert post.text == text
        assert isinstance(post.text, Text)

    def test_missing_text_should_raise_error(self) -> None:
        with pytest.raises(ValueError, match="textは必須です。"):
            Post(text=None) # type: ignore[arg-type]

    def test_default_id_is_zero(self) -> None:
        text = Text("hello")
        post = Post(text=text)

        assert post.id == 0

    def test_from_raw_should_skip_validation(self) -> None:
        raw_text = Text.from_raw("raw-text")
        post = Post.from_raw(raw_id=123, raw_text=raw_text)

        assert isinstance(post, Post)
        assert post.id == 123
        assert post.text == raw_text
        assert post.text.value == "raw-text"

・「src/tests/application/usecase/post/test_create_post.py」

import pytest
from pytest_mock import MockerFixture

from myapp.application.usecase.post import CreatePostUsecase
from myapp.domain.post import Post, Text
from myapp.presentation.schema.post import PostCreateRequest, PostResponse


@pytest.mark.unit
@pytest.mark.asyncio
class TestCreatePost:
    # 正常系テスト
    async def test_create_post_success(self, mocker: MockerFixture) -> None:
        # ===================
        # モック作成
        # ===================
        mock_logger = mocker.Mock()
        mock_db = mocker.Mock()
        mock_post_repo = mocker.Mock()

        # リポジトリのモック化
        mock_post_return_value = Post(text=Text("Postデータ"))
        mock_post_return_value.id = 1
        mock_post_repo.create = mocker.AsyncMock(return_value=mock_post_return_value)

        # ユースケースのモック化
        usecase = CreatePostUsecase(
            logger=mock_logger, db=mock_db, post_repo=mock_post_repo
        )

        # ===================
        # リクエストデータ作成
        # ===================
        req = PostCreateRequest(text="Postデータ")

        # ===================
        # テスト実行
        # ===================
        res = await usecase.execute(req)

        # ===================
        # 検証
        # ===================
        assert isinstance(res, PostResponse)
        assert res.id == 1
        assert res.text == "Postデータ"

        # 対象のリポジトリが1回呼ばれること
        mock_post_repo.create.assert_awaited_once()

※非同期関数の場合は「@pytest.mark.asyncio」を付けます。ユニットテストでは各種モック化してテストします。

・「src/tests/application/usecase/post/test_get_posts.py」

import pytest
from pytest_mock import MockerFixture

from myapp.application.usecase.post import GetPostsUsecase
from myapp.domain.post import Post, Text
from myapp.presentation.schema.post import PostResponse


@pytest.mark.unit
@pytest.mark.asyncio
class TestGetPosts:
    # 正常系テスト
    async def test_get_posts_success(self, mocker: MockerFixture) -> None:
        # ===================
        # モック作成
        # ===================
        mock_logger = mocker.Mock()
        mock_db = mocker.Mock()
        mock_post_repo = mocker.Mock()

        # リポジトリのモック化
        mock_post_return_value = [
            Post.from_raw(raw_id=1, raw_text=Text.from_raw(raw_value="Postデータ１")),
            Post.from_raw(raw_id=2, raw_text=Text.from_raw(raw_value="Postデータ２")),
        ]
        mock_post_repo.find_all = mocker.AsyncMock(return_value=mock_post_return_value)

        # ユースケースのモック化
        usecase = GetPostsUsecase(
            logger=mock_logger, db=mock_db, post_repo=mock_post_repo
        )

        # ===================
        # テスト実行
        # ===================
        res = await usecase.execute()

        # ===================
        # 検証
        # ===================
        assert isinstance(res, list)
        assert all(isinstance(item, PostResponse) for item in res)
        assert res[0].id == 1
        assert res[0].text == "Postデータ１"
        assert res[1].id == 2
        assert res[1].text == "Postデータ２"

        # 対象のリポジトリが1回呼ばれること
        mock_post_repo.find_all.assert_awaited_once()

・「src/tests/presentation/handler/post/test_post.py」

import pytest
from fastapi.testclient import TestClient

from myapp.main import app

client = TestClient(app)


@pytest.mark.integration
class TestPostAPI:
    def test_create_post_success(self) -> None:
        # テスト実行
        res = client.post(
            "/post",
            json={"text": "Postテキスト"},
        )

        # 検証
        assert res.status_code == 201
        assert res.json() == {"id": 0, "text": "Postテキスト"}

    def test_create_post_valid_error(self) -> None:
        # テスト実行
        res = client.post(
            "/post",
            json={"text": "aaaaabbbbbcccccddddd1"},
        )

        # 検証
        assert res.status_code == 422
        data = res.json()
        assert "String should have at most 20 characters" in data["detail"][0]["msg"]
        assert data["detail"][0]["ctx"]["max_length"] == 20

    def test_get_posts_success(self) -> None:
        # テスト実行
        res = client.get("/posts")

        # 検証
        assert res.status_code == 200
        assert res.json() == [
            {"id": 1, "text": "Postデータ１"},
            {"id": 2, "text": "Postデータ２"},
        ]

※ハンドラーのテストはインテグレーションテストとして、リクエストを実行してテストします。（もし実際にDB操作がある場合はそれも実行されるため、テスト用のDB設定も必要になります。）

次に以下のコマンドを実行し、ユニットテストを実行します。

$ docker compose exec fastapi poetry run pytest -m unit

コマンドを実行後、以下のように全てのテストがPASSすればOKです。

※オプション「-m unit」でマーク「unit」が付いたものだけ実行しているため、それ以外のスキップしたテストが「3 deselected」として表示されますが、それは仕様なので気にしなくてOKです。

次に以下のコマンドを実行し、インテグレーションテストを実行します。

$ docker compose exec fastapi poetry run pytest -m integration

コマンドを実行後、以下のように全てのテストがPASSすればOKです。

データベースついて

今回はDB部分は省略しましたが、組み込みたい場合は以前に書いた以下の記事などを参考にしつつ（生成AIを使っていない時代に書いた記事で情報が古いので参考程度にどうぞ）、必要な部分を追加修正していけばできると思います。

関連記事👇

【FastAPI入門】Dockerで環境構築してPythonのAPIを開発する方法まとめ

2024年5月4日

あるドメインのデータを別のドメインで使いたい場合について

上記では基本的なDDD構成についてご紹介しましたが、実際にはあるドメインのデータを別のドメインのユースケースなどで使いたいというようなことも出てくると思います。

その際はどう処理を書くべきか悩ましいところですが、トランザクション管理（データベースで「安全にまとめて作業をする仕組み」のこと）の観点からすると1ハンドラー1ユースケースで作るのが最適なため、一つのユースケース内で各種ドメインを呼び出すような形で作って下さい。

ただし、将来的にマイクロサービス化したくなった場合、ドメイン単位で独立した設計になっていないユースケースの部分については分割するのが難しくなりますが、その点はしょうがないのであきらめましょう！

※ただし、ドメインの境界線はしっかり分けてDB設計、データ更新していくのが大事ではあるので、その点はモノリスになりすぎないように注意して下さい。

本番環境用Dockerコンテナを作って試す

次に本番環境へのデプロイを想定し、専用のDockerコンテナを作ってローカル環境で試してみますが、コンテナ一つで複数ワーカーを動かしたいかどうかで若干やり方が変わるため、それぞれご紹介します。

まずは以下のコマンドを実行し、本番環境用のDockerfileを作成します。

$ mkdir -p deploy/docker/prod && touch deploy/docker/prod/Dockerfile

k8sやCloud Runなどコンテナ単位でスケールする場合

k8sやCloud Runなどコンテナ単位でスケールする場合、上記までのローカル環境と同様に「Uvicorn」のみで起動させればよいです。

その場合、ファイル「deploy/docker/prod/Dockerfile」は以下のように記述します。

・「deploy/docker/prod/Dockerfile」

####################
# ビルドステージ
####################
FROM python:3.14.0-slim-trixie AS builder

# poetryをインストール
RUN pip3 install --no-cache-dir poetry

# プラグイン「poetry-plugin-export」を追加
RUN poetry self add poetry-plugin-export

WORKDIR /build

# 依存関係のみコピー
COPY pyproject.toml poetry.lock ./

# 本番環境用の依存関係のみをファイル「requirements.txt」に記述
RUN poetry export -f requirements.txt --without-hashes --only main -o requirements.txt

####################
# 実行ステージ
####################
FROM python:3.14.0-slim-trixie AS runner

# 環境変数設定
ENV ENV=production

# テスト用の環境変数設定（ローカルではコマンドで渡す必要があるため）
ARG TEST_VALUE
ENV TEST_VALUE=${TEST_VALUE}

# タイムゾーン設定
ENV TZ=Asia/Tokyo

WORKDIR /py

# 必要なパッケージをインストール
COPY --from=builder /build/requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt

COPY ./src/myapp ./src/myapp

# 非rootユーザーを設定
RUN addgroup --system --gid 1001 appuser && \
    adduser --system --uid 1001 appuser
USER appuser

EXPOSE 9004

# k8sやCloud Runなどコンテナ単位でスケールする場合
# Uvicornのみで起動させる
CMD ["uvicorn", "myapp.main:app", "--app-dir", "src", "--host", "0.0.0.0", "--port", "9004"]

※環境変数「TEST_VALUE」は必須にしているため、ビルドコマンド実行時に値を渡せるようにしています。

ECSなどでコンテナ一つで複数ワーカーを動かしたい場合

ECSなどでコンテナ一つで複数ワーカーを動かしたい場合、サーバー起動には「Gunicorn + UvicornWorker」の組み合わせを利用します。

その場合、まずは以下のコマンドを実行し、poetryで「gunicorn」をインストールします。

$ docker compose exec fastapi poetry add gunicorn

次に以下のコマンドを実行し、Gunicorn用の設定ファイルを作成します。

touch gunicorn_conf.py

次に作成したファイルを以下のように記述します。

・「gunicorn_conf.py」

import os

# サーバー設定
bind = "0.0.0.0:9004"
workers = int(os.getenv("WORKERS", "4"))
worker_class = "uvicorn.workers.UvicornWorker"

# ログ設定
loglevel = "info"
accesslog = "-" # 標準出力に出す
errorlog = "-" # 標準エラー出力に出す

次に事前に作成したファイル「deploy/docker/prod/Dockerfile」は以下のように記述します。

・「deploy/docker/prod/Dockerfile」

####################
# ビルドステージ
####################
FROM python:3.14.0-slim-trixie AS builder

# poetryをインストール
RUN pip3 install --no-cache-dir poetry

# プラグイン「poetry-plugin-export」を追加
RUN poetry self add poetry-plugin-export

WORKDIR /build

# 依存関係のみコピー
COPY pyproject.toml poetry.lock ./

# 本番環境用の依存関係のみをファイル「requirements.txt」に記述
RUN poetry export -f requirements.txt --without-hashes --only main -o requirements.txt

####################
# 実行ステージ
####################
FROM python:3.14.0-slim-trixie AS runner

# 環境変数設定
ENV ENV=production

# テスト用の環境変数設定（ローカルではコマンドで渡す必要があるため）
ARG TEST_VALUE
ENV TEST_VALUE=${TEST_VALUE}

# タイムゾーン設定
ENV TZ=Asia/Tokyo

WORKDIR /py

# 必要なパッケージをインストール
COPY --from=builder /build/requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt

COPY ./src/myapp ./src/myapp

# 非rootユーザーを設定
RUN addgroup --system --gid 1001 appuser && \
    adduser --system --uid 1001 appuser
USER appuser

EXPOSE 9004

# ECSなどでコンテナ一つで複数ワーカーを動かしたい場合
# （Gunicorn + UvicornWorker）の組み合わせ
COPY gunicorn_conf.py .
WORKDIR /py/src
CMD ["gunicorn", "-c", "/py/gunicorn_conf.py", "myapp.main:app"]

※環境変数「TEST_VALUE」は必須にしているため、ビルドコマンド実行時に値を渡せるようにしています。

Dockerコンテナ単体でのビルドと起動

次に以下のコマンドを実行し、Dockerコンテナのビルドおよび起動をします。

$ docker compose down
$ docker build --no-cache --build-arg TEST_VALUE='PROD-VALUE' -f ./deploy/docker/prod/Dockerfile -t fastapi-domain-api:latest .
$ docker run -d -p 80:9004 fastapi-domain-api:latest

※ビルド時に「–build-arg TEST_VALUE=’PROD-VALUE’」で環境変数への値を渡しています。今回はテストなのでタグは「latest」ですが、実際にはバージョンのタグを指定するのでご注意下さい。

PostmanでAPIを試す

次にエンドポイントを「http://localhost/」とし、上記と同様にAPIをPostmanで実行して試して下さい。

各種APIを実行し、想定通りの結果になればOKです。

最後に

今回はPythonのFastAPIでDDD構成のバックエンドAPIを開発する方法について解説しました。

これから数年間は間違いなくAI関連機能の開発が盛り上がるのは避けられないため、その際は主にPythonを使ったAPI開発が増えるのではと思います。

その際は軽量でバランスがいいFastAPIでのAPI開発が増えそうですが、実務ではドメイン駆動設計で開発することが多いと思うので、FastAPIでDDD構成のバックエンドAPIを開発したい方はぜひ参考にしてみて下さい。

この記事を書いた人
最新の記事

Tomoyuki

SEを5年経験後、全くの未経験ながら思い切ってブロガーに転身し、月間13万PVを達成。その後コロナの影響も受け、以前から興味があったWeb系エンジニアへのキャリアチェンジを決意。現在はWeb系エンジニアとして働きながら、プロゲーマーとしても活躍できるように活動中。

カテゴリー

【2025年】おすすめのゲーミングPC

PythonのFastAPIでDDD（ドメイン駆動設計）構成のバックエンドAPIを開発する方法まとめ