DDD (ドメイン駆動設計) アーキテクチャ

概要

本プロジェクトは、ドメイン駆動設計（DDD）をベースにしたクリーンアーキテクチャを採用しています。これにより、ビジネスロジックの純粋性を保ちながら、技術的な詳細から分離された保守性の高いシステムを実現します。

アーキテクチャの全体像

graph TD
    subgraph Interfaces層 [Interfaces 層（FastAPI）]
        A1[REST API / GraphQL / CLI]
        A2[Request DTO / Response DTO]
    end

    subgraph Application層 [Application 層（ユースケース）]
        B1[UserSummaryUseCase]
        B2[複数DB統合ロジック]
    end

    subgraph Domain層 [Domain 層（ビジネスルール）]
        C1[Entity: UserProfile]
        C2[Value Object]
        C3[Domain Service]
    end

    subgraph Infrastructure層 [Infrastructure 層（技術詳細）]
        D1[PostgreSQL Repository]
        D2[Neo4j Repository]
        D3[External API Client]
    end

    A1 --> A2
    A2 --> B1
    B1 --> C1
    B1 --> D1
    B1 --> D2
    D1 --> C1
    D2 --> C1

依存関係の方向

graph LR
    Interfaces --> Application
    Application --> Domain
    Infrastructure --> Domain
    Interfaces -.-> Infrastructure

    style Domain fill:#90EE90
    style Application fill:#87CEEB
    style Infrastructure fill:#FFB6C1
    style Interfaces fill:#FFD700

依存の原則

• 内向き依存: すべての依存は内側（Domain層）に向かう
• Domain層: 外部への依存なし（純粋なビジネスロジック）
• Application層: Domainに依存、Infrastructureは抽象に依存
• Infrastructure層: Domainの抽象（Repository Interface）に依存
• Interfaces層: Applicationを呼び出し、Infrastructureを注入

各層の責務

1. Domain層（ドメイン層）

責務: ビジネスルールとビジネスロジックの中核

含まれるもの: - Entity（エンティティ）: ビジネス上の重要な概念（例: User, Order） - Value Object（値オブジェクト）: 不変の値（例: Email, Money） - Domain Service（ドメインサービス）: 複数エンティティにまたがるビジネスロジック - Repository Interface（リポジトリインターフェイス）: データ永続化の抽象

特徴: - 外部ライブラリへの依存なし（pure Python） - フレームワーク非依存 - テストが最も容易

実装例:

# app/domain/user.py
from dataclasses import dataclass
from datetime import datetime

@dataclass
class UserProfile:
    """ユーザープロファイルエンティティ"""
    id: int
    name: str
    email: str
    joined_at: datetime

    def is_premium_user(self) -> bool:
        """プレミアムユーザー判定ロジック"""
        days_since_joined = (datetime.now() - self.joined_at).days
        return days_since_joined > 30

    def to_dict(self) -> dict:
        """辞書形式に変換"""
        return {
            "id": self.id,
            "name": self.name,
            "email": self.email,
            "joined_at": self.joined_at.isoformat(),
            "is_premium": self.is_premium_user()
        }

# app/domain/repository_interface.py
from abc import ABC, abstractmethod
from typing import Optional
from app.domain.user import UserProfile

class UserRepositoryInterface(ABC):
    """ユーザーリポジトリの抽象"""

    @abstractmethod
    def find_by_id(self, user_id: int) -> Optional[UserProfile]:
        pass

    @abstractmethod
    def save(self, user: UserProfile) -> UserProfile:
        pass

2. Application層（アプリケーション層・ユースケース層）

責務: ビジネスユースケースの調整・統合

含まれるもの: - UseCase（ユースケース）: ビジネスシナリオの実装 - 複数DBの統合ロジック: 異なるリポジトリからのデータ結合 - トランザクション管理: 複数の操作をまとめる

特徴: - Domainのエンティティとサービスを組み合わせる - Repositoryインターフェイスを通じてデータアクセス - ビジネスフローの調整役

実装例:

# app/application/user_summary_usecase.py
from app.domain.user import UserProfile
from app.domain.repository_interface import UserRepositoryInterface

class UserSummaryUseCase:
    """ユーザーサマリー取得ユースケース"""

    def __init__(
        self,
        corporate_repo: UserRepositoryInterface,
        auth_repo: UserRepositoryInterface
    ):
        self.corporate_repo = corporate_repo
        self.auth_repo = auth_repo

    def execute(self, user_id: int) -> dict:
        """
        複数DBからユーザー情報を統合して返す

        Args:
            user_id: ユーザーID

        Returns:
            統合されたユーザー情報
        """
        # corporateDBからプロファイル取得
        profile = self.corporate_repo.find_by_id(user_id)
        if not profile:
            raise ValueError(f"User {user_id} not found")

        # auth_sharedDBから認証情報取得
        auth_info = self.auth_repo.find_by_id(user_id)

        # 統合してJSONに変換
        return {
            "user_id": profile.id,
            "name": profile.name,
            "email": auth_info.email if auth_info else profile.email,
            "joined_at": profile.joined_at.isoformat(),
            "is_premium": profile.is_premium_user()
        }

3. Infrastructure層（インフラストラクチャ層）

責務: 技術的な詳細の実装

含まれるもの: - Repository実装: DB接続の具体的実装 - 外部API Client: 外部サービスとの通信 - ORM Model: SQLAlchemyなどのORMモデル

特徴: - Domainのインターフェイスを実装 - DB接続先ごとに分離（corporate / miniapp_lab / auth_shared） - 技術スタックの変更が容易

マルチDB構成:

infrastructure/
├─ corporate/
│   ├─ repository.py          # corporate DB用リポジトリ実装
│   ├─ orm_model.py           # corporate DB用ORMモデル
│   └─ connection.py          # corporate DB接続設定
├─ miniapp_lab/
│   ├─ repository.py
│   ├─ orm_model.py
│   └─ connection.py
├─ auth_shared/
│   ├─ repository.py
│   ├─ orm_model.py
│   └─ connection.py
└─ vector/
    ├─ pgvector_repository.py  # ベクトル検索用
    └─ vector_model.py

実装例:

# app/infrastructure/corporate/repository.py
from sqlalchemy.orm import Session
from app.domain.user import UserProfile
from app.domain.repository_interface import UserRepositoryInterface
from app.infrastructure.corporate.orm_model import UserORM

class CorporateUserRepository(UserRepositoryInterface):
    """Corporate DB用ユーザーリポジトリ実装"""

    def __init__(self, session: Session):
        self.session = session

    def find_by_id(self, user_id: int) -> Optional[UserProfile]:
        """IDでユーザーを取得"""
        user_orm = self.session.query(UserORM).filter(
            UserORM.id == user_id
        ).first()

        if not user_orm:
            return None

        # ORMモデルからドメインエンティティに変換
        return UserProfile(
            id=user_orm.id,
            name=user_orm.name,
            email=user_orm.email,
            joined_at=user_orm.created_at
        )

    def save(self, user: UserProfile) -> UserProfile:
        """ユーザーを保存"""
        user_orm = UserORM(
            id=user.id,
            name=user.name,
            email=user.email,
            created_at=user.joined_at
        )
        self.session.add(user_orm)
        self.session.commit()
        return user

# app/infrastructure/corporate/orm_model.py
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class UserORM(Base):
    """Corporate DB用ユーザーORMモデル"""
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(255), nullable=False)
    email = Column(String(255), nullable=False)
    created_at = Column(DateTime, nullable=False)

4. Interfaces層（インターフェイス層）

責務: 外部とのやり取り（入出力）

含まれるもの: - API Router: FastAPIのエンドポイント定義 - DTO (Data Transfer Object): リクエスト/レスポンスの型定義 - Dependency Injection: 依存性注入の設定

特徴: - Applicationレイヤーのユースケースを呼び出す - HTTPリクエストをDTOに変換 - Pydanticでバリデーション

実装例:

# app/interfaces/routers/user_router.py
from fastapi import APIRouter, Depends
from app.application.user_summary_usecase import UserSummaryUseCase
from app.interfaces.dependencies import get_user_summary_usecase
from app.interfaces.schemas.user_schema import UserSummaryResponse

router = APIRouter(prefix="/api/users", tags=["users"])

@router.get("/{user_id}", response_model=UserSummaryResponse)
async def get_user_summary(
    user_id: int,
    usecase: UserSummaryUseCase = Depends(get_user_summary_usecase)
):
    """
    ユーザーサマリーを取得

    複数DBから情報を統合して返します
    """
    result = usecase.execute(user_id)
    return result

# app/interfaces/schemas/user_schema.py
from pydantic import BaseModel
from datetime import datetime

class UserSummaryResponse(BaseModel):
    """ユーザーサマリーレスポンス"""
    user_id: int
    name: str
    email: str
    joined_at: str
    is_premium: bool

    class Config:
        json_schema_extra = {
            "example": {
                "user_id": 1,
                "name": "田中太郎",
                "email": "tanaka@example.com",
                "joined_at": "2024-01-01T00:00:00",
                "is_premium": True
            }
        }

# app/interfaces/dependencies.py
from fastapi import Depends
from sqlalchemy.orm import Session
from app.application.user_summary_usecase import UserSummaryUseCase
from app.infrastructure.corporate.repository import CorporateUserRepository
from app.infrastructure.auth_shared.repository import AuthUserRepository
from app.core.database import get_corporate_db, get_auth_db

def get_user_summary_usecase(
    corporate_db: Session = Depends(get_corporate_db),
    auth_db: Session = Depends(get_auth_db)
) -> UserSummaryUseCase:
    """ユースケースの依存性注入"""
    corporate_repo = CorporateUserRepository(corporate_db)
    auth_repo = AuthUserRepository(auth_db)
    return UserSummaryUseCase(corporate_repo, auth_repo)

データフロー（シーケンス図）

sequenceDiagram
    participant UI as FastAPI Controller (Interfaces)
    participant UC as UseCase (Application)
    participant DM as Domain Entity
    participant RP1 as CorporateRepository (Infrastructure)
    participant RP2 as AuthRepository (Infrastructure)
    participant DB1 as PostgreSQL (corporate)
    participant DB2 as PostgreSQL (auth_shared)

    UI->>UC: リクエスト受信 (user_id)
    UC->>RP1: find_by_id(user_id)
    RP1->>DB1: SQL実行
    DB1-->>RP1: データ返却
    RP1->>DM: ORMからEntityに変換
    DM-->>UC: UserProfile返却

    UC->>RP2: find_by_id(user_id)
    RP2->>DB2: SQL実行
    DB2-->>RP2: データ返却
    RP2->>DM: ORMからEntityに変換
    DM-->>UC: AuthInfo返却

    UC->>UC: データ統合・ビジネスロジック適用
    UC-->>UI: 統合結果を返却 (JSON)
    UI-->>Client: HTTPレスポンス

マルチデータベース統合パターン

3つのデータベース構成

graph TB
    subgraph Application Layer
        UC[UserSummaryUseCase]
    end

    subgraph Infrastructure Layer
        R1[CorporateRepository]
        R2[MiniAppRepository]
        R3[AuthRepository]
    end

    subgraph Database Layer
        DB1[(PostgreSQL<br/>corporate)]
        DB2[(PostgreSQL<br/>miniapp_lab)]
        DB3[(PostgreSQL<br/>auth_shared)]
    end

    UC --> R1
    UC --> R2
    UC --> R3
    R1 --> DB1
    R2 --> DB2
    R3 --> DB3

    style UC fill:#87CEEB
    style R1 fill:#FFB6C1
    style R2 fill:#FFB6C1
    style R3 fill:#FFB6C1
    style DB1 fill:#98FB98
    style DB2 fill:#98FB98
    style DB3 fill:#98FB98

データベース別の役割

データベース	用途	主なテーブル
corporate	コーポレートサイト用データ	users, posts, pages
miniapp_lab	小規模実験サービス	experiments, prototypes
auth_shared	認証・ユーザー情報共通化	auth_users, sessions, permissions

統合処理の責務分離

処理内容	責務を持つ層	理由
DBごとのデータ取得	Infrastructure層（Repository）	永続化層の責務
取得したデータの結合や整形	Application層（UseCase）	ユースケース固有の組み合わせロジック
ビジネスルールの適用	Domain層（Entity/Service）	ビジネスロジックの純粋性保持
JSONへの変換	Application層 → Interfaces層	出力形式の制御

テスト戦略

レイヤー別のテストアプローチ

graph LR
    subgraph テスト種別
        T1[Unit Test<br/>ユニットテスト]
        T2[Integration Test<br/>統合テスト]
        T3[E2E Test<br/>エンドツーエンドテスト]
    end

    subgraph 対象レイヤー
        L1[Domain層]
        L2[Application層]
        L3[Infrastructure層]
        L4[Interfaces層]
    end

    T1 --> L1
    T1 --> L2
    T2 --> L3
    T3 --> L4

テスト対象	テスト種類	概要	ツール
Domain層	ユニットテスト	pure Python。外部依存なし	pytest
Application層	ユースケーステスト	モックリポジトリでDBを外す	pytest + unittest.mock
Infrastructure層	統合テスト	Dockerコンテナ上で実DB接続	pytest + testcontainers
Interfaces層	APIテスト	FastAPIのTestClientを使用	pytest + TestClient

TDD（テスト駆動開発）の流れ

graph TD
    A[1. ユースケーステストを書く] --> B[2. Domainエンティティ設計]
    B --> C[3. Repositoryインターフェイス定義]
    C --> D[4. モックでテスト実行]
    D --> E{テスト成功?}
    E -->|No| B
    E -->|Yes| F[5. Infrastructure実装]
    F --> G[6. 統合テスト実行]
    G --> H{テスト成功?}
    H -->|No| F
    H -->|Yes| I[7. Interfaces層実装]
    I --> J[8. E2Eテスト実行]
    J --> K[完了]

ディレクトリ構成

project-root/
├─ app/
│   ├─ main.py                # FastAPI起動
│   ├─ core/                  # 共通設定
│   │   ├─ config.py          # 設定管理
│   │   ├─ database.py        # DB接続管理
│   │   └─ logging.py         # ログ設定
│   ├─ domain/                # ドメイン層
│   │   ├─ user.py            # Userエンティティ
│   │   ├─ value_objects.py   # 値オブジェクト
│   │   ├─ domain_service.py  # ドメインサービス
│   │   └─ repository_interface.py  # リポジトリIF
│   ├─ application/           # アプリケーション層
│   │   ├─ user_summary_usecase.py
│   │   └─ create_user_usecase.py
│   ├─ infrastructure/        # インフラストラクチャ層
│   │   ├─ corporate/         # corporate DB用
│   │   │   ├─ repository.py
│   │   │   ├─ orm_model.py
│   │   │   └─ connection.py
│   │   ├─ miniapp_lab/       # miniapp DB用
│   │   │   ├─ repository.py
│   │   │   └─ orm_model.py
│   │   ├─ auth_shared/       # auth DB用
│   │   │   ├─ repository.py
│   │   │   └─ orm_model.py
│   │   └─ vector/            # pgvector用
│   │       ├─ repository.py
│   │       └─ model.py
│   ├─ interfaces/            # インターフェイス層
│   │   ├─ routers/           # APIルーター
│   │   │   ├─ user_router.py
│   │   │   └─ health_router.py
│   │   ├─ schemas/           # Pydantic DTO
│   │   │   ├─ user_schema.py
│   │   │   └─ response_schema.py
│   │   └─ dependencies.py    # DI設定
│   └─ tests/                 # テスト
│       ├─ unit/              # ユニットテスト
│       ├─ integration/       # 統合テスト
│       └─ e2e/               # E2Eテスト
├─ docs/                      # ドキュメント
├─ docker-compose.yaml
└─ .env

まとめ

設計原則

項目	内容
設計思想	DDDをベースにしたクリーンアーキテクチャ
依存方向	内向き（Domainが最も純粋）
責務分離	Router=入出力、Application=統合、Domain=ビジネス
DB複数接続	Infrastructureで分離、Applicationで統合
テスト	Domain=ユニット、Application=モック、Infrastructure=統合テスト

利点

1. ビジネスロジックの純粋性: Domainが外部依存を持たない
2. テスタビリティ: 各層を独立してテスト可能
3. 保守性: 責務が明確で変更の影響範囲が限定的
4. 拡張性: 新しいDBやサービスの追加が容易
5. 技術スタックの変更: InfrastructureとInterfacesを変えるだけ

適用シーン

• 長期運用が想定されるプロジェクト
• チーム開発（責務分離により並行開発が容易）
• 複雑なビジネスルールを持つシステム
• 複数のデータソースを統合する必要がある場合
• テストカバレッジを高く保ちたい場合