DDDを適用したレイヤードアーキテクチャとは？Pythonサンプルでわかりやすく解説！

myapp-ddd/               ← プロジェクトのルート
├── main.py              ← エントリーポイント（Flask起動 & 依存注入）
└── myapp/               ← アプリケーション本体
    ├── controller/       ← コントローラー層（外部I/Oとの接点）
    │   └── user_controller.py
    ├── usecase/          ← ユースケース層（アプリケーションのシナリオ）
    │   └── user_usecase.py
    ├── domain/           ← ドメイン層（ビジネスルール）
    │   ├── user.py
    │   ├── user_repository.py
    │   └── user_service.py
    └── infrastructure/   ← インフラストラクチャ層（技術的な処理）
        └── user_repository_impl.py

from flask import Flask, request, jsonify
from myapp.usecase.user_usecase import UserUseCase

app = Flask(__name__)

# main.py から依存注入される
user_usecase: UserUseCase = None


@app.route("/users", methods=["POST"])
def create_user():
    """
    ユーザー登録API
    - ユーザーからの入力は JSON (文字列形式)
    - ユースケース層では int や ValueObject を使う
    - コントローラーで型変換・マッピングを行う
    """
    data = request.json

    try:
        # 入力マッピング (string → int)
        user_id = int(data["id"])
        email = data["email"]
        name = data["name"]

        # ユースケース呼び出し
        user_usecase.register(user_id, email, name)

        # 結果を JSON 形式に整形して返す
        return jsonify({
            "status": "success",
            "message": f"ユーザー登録成功: id={user_id}, email={email}, name={name}"
        }), 201

    except Exception as e:
        return jsonify({
            "status": "error",
            "message": str(e)
        }), 400


@app.route("/users/<user_id>", methods=["GET"])
def show_user(user_id):
    """
    ユーザー情報取得API
    - URLパラメータは文字列 (str)
    - ユースケース層では int を使うため変換する
    """
    try:
        user = user_usecase.fetch_user(int(user_id))
        if user:
            # ドメイン層のオブジェクト → JSON にマッピング
            return jsonify({
                "id": user.user_id,
                "name": user.name,
                "email": user.email.address
            }), 200
        else:
            return jsonify({"message": "ユーザーが見つかりません"}), 404

    except Exception as e:
        return jsonify({"status": "error", "message": str(e)}), 400

from myapp.domain.user_service import UserService
from myapp.domain.user import User, Email
from typing import Optional

class UserUseCase:
    """アプリケーションのユースケースを実現する層"""

    def __init__(self, user_service: UserService):
        self.user_service = user_service

    def register(self, user_id: int, email: str, name: str) -> None:
        """
        ユーザー登録のユースケースを実行する。
        - すでにユーザーが存在するか確認
        - 存在しなければ新規登録
        """
        existing_user = self.user_service.get_user(user_id)
        if existing_user:
            raise ValueError("このユーザーIDはすでに登録されています")

        self.user_service.register_user(user_id, email, name)

    def fetch_user(self, user_id: int) -> Optional[User]:
        """
        ユーザー情報取得のユースケースを実行する。
        """
        return self.user_service.get_user(user_id)

from dataclasses import dataclass

# Value Object: Email
@dataclass(frozen=True)
class Email:
    """メールアドレスを表す値オブジェクト（不変）"""
    address: str

    def __post_init__(self):
        if "@" not in self.address:
            raise ValueError("Invalid email address")


# Entity: User
class User:
    """ユーザーを表すエンティティ（一意なIDを持つ）"""

    def __init__(self, user_id: int, email: Email, name: str):
        self.user_id = user_id  # 識別子
        self.email = email      # 値オブジェクトを属性に持つ
        self.name = name

    def change_email(self, new_email: Email):
        """ユーザーのメールアドレスを変更する"""
        self.email = new_email

    def __repr__(self):
        return f"User(id={self.user_id}, name={self.name}, email={self.email.address})"

from abc import ABC, abstractmethod
from typing import Optional
from myapp.domain.user import User


class UserRepository(ABC):
    """
    ユーザー情報の永続化操作を定義するリポジトリインターフェース。
    具体的な実装はインフラストラクチャ層に委ねる。
    """

    @abstractmethod
    def find_by_id(self, user_id: int) -> Optional[User]:
        """
        指定のユーザーIDに該当するユーザーを返す。

        Args:
            user_id (int): 検索対象のユーザーID。

        Returns:
            Optional[User]: ユーザーが存在する場合はUserオブジェクト、存在しない場合はNone。
        """
        pass

    @abstractmethod
    def save(self, user: User) -> None:
        """
        ユーザー情報を保存または更新する。

        Args:
            user (User): 保存したいユーザーオブジェクト。

        Returns:
            None
        """
        pass

from typing import Optional
from myapp.domain.user_repository import UserRepository
from myapp.domain.user import User, Email

class UserService:
    """
    ドメインサービス。
    - エンティティ単体に置くと不自然または肥大化する「ルール」をまとめる。
    - リポジトリの「抽象」に依存して確認・保存を指示する（実装詳細は知らない）。
    """

    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository

    def register_user(self, user_id: int, email_str: str, name: str) -> None:
        """
        ユーザー登録のルールを適用してユーザーを作成・保存する。

        流れ:
          1) メール文字列から Email 値オブジェクトを生成（形式チェックは Email 側）
          2) ID の重複有無をリポジトリ経由で確認
          3) 問題なければ User を生成し、保存を指示（実装はインフラ層）

        Raises:
            ValueError: ID が既に使われている場合
        """
        email = Email(email_str)  # 値オブジェクト側で形式バリデーション
        existing = self.user_repository.find_by_id(user_id)
        if existing is not None:
            raise ValueError("User with this ID already exists")

        user = User(user_id, email, name)
        self.user_repository.save(user)

    def get_user(self, user_id: int) -> Optional[User]:
        """
        ユーザーIDでユーザーを取得する（保存実装は知らない／委譲する）。
        """
        return self.user_repository.find_by_id(user_id)

import sqlite3
from typing import Optional
from myapp.domain.user_repository import UserRepository
from myapp.domain.user import User, Email

class SQLiteUserRepository(UserRepository):
    """SQLiteを使ったユーザーリポジトリ実装"""

    def __init__(self, db_path: str = "myapp.db"):
        self.db_path = db_path
        self._initialize_db()

    def _initialize_db(self):
        """テーブルがなければ作成"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                email TEXT NOT NULL,
                name TEXT NOT NULL
            )
        """)
        conn.commit()
        conn.close()

    def find_by_id(self, user_id: int) -> Optional[User]:
        """ユーザーIDで検索"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("SELECT id, email, name FROM users WHERE id = ?", (user_id,))
        row = cursor.fetchone()
        conn.close()

        if row:
            return User(user_id=row[0], email=Email(row[1]), name=row[2])
        return None

    def save(self, user: User) -> None:
        """ユーザーを保存（INSERT or REPLACE）"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            INSERT OR REPLACE INTO users (id, email, name)
            VALUES (?, ?, ?)
        """, (user.user_id, user.email.address, user.name))
        conn.commit()
        conn.close()

from myapp.controller import user_controller
from myapp.usecase.user_usecase import UserUseCase
from myapp.domain.user_service import UserService
from myapp.infrastructure.user_repository_impl import SQLiteUserRepository

# 依存関係を組み立てる
repo = SQLiteUserRepository("myapp.db")
service = UserService(repo)
usecase = UserUseCase(service)

# Flaskコントローラに依存を注入
user_controller.user_usecase = usecase

if __name__ == "__main__":
    user_controller.app.run(debug=True, host="0.0.0.0", port=5000)

# 仮想環境を作る（初回のみ）
python3 -m venv .venv

# 仮想環境に入る
source .venv/bin/activate

pip install flask

python main.py

(.venv) user01@Ubuntu:~/work/myapp-ddd$ python main.py
 * Serving Flask app 'myapp.controller.user_controller'
 * Debug mode: on
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on all addresses (0.0.0.0)
 * Running on http://127.0.0.1:5000
 * Running on http://172.27.236.162:5000
Press CTRL+C to quit
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 294-647-968

curl -X POST -H "Content-Type: application/json" \
  -d '{"id":"1","email":"test@example.com","name":"太郎"}' \
  http://127.0.0.1:5000/users

{
  "status": "success",
  "message": "ユーザー登録成功: id=1, email=test@example.com, name=太郎"
}

curl http://127.0.0.1:5000/users/1

{
  "id": 1,
  "name": "太郎",
  "email": "test@example.com"
}