English | 日本語
- Go実装版: oniongo
注意: このリポジトリは「PythonのWebアプリケーションでDDDアーキテクチャを実装する方法」を説明するためのサンプルです。参考として使用する場合は、本番環境にデプロイする前に認証とセキュリティの実装を追加してください。
- DeepWiki powered by Devin: https://deepwiki.com/iktakahiro/dddpy
- ブログ記事: Python DDD オニオンアーキテクチャ
- Python 3.13 以上
- uv - Pythonパッケージインストーラー＆リゾルバー
- uvを使用して依存関係をインストールします：
make install- Webアプリケーションを実行します：
make devこのリポジトリのソフトウェアアーキテクチャは、オニオン・アーキテクチャに基づいています。
オニオン・アーキテクチャを実践するためのディレクトリ構造に正解はありませんが、このリポジトリでは次のようなディレクトリ構造を採用しています：
├── main.py
├── dddpy
│ ├── domain
│ │ └── todo
│ │ ├── entities
│ │ │ └── todo.py
│ │ ├── value_objects
│ │ │ ├── todo_title.py
│ │ │ ├── todo_description.py
│ │ │ ├── todo_id.py
│ │ │ └── todo_status.py
│ │ ├── repositories
│ │ │ └── todo_repository.py
│ │ └── exceptions
│ ├── infrastructure
│ │ ├── di
│ │ │ └── injection.py
│ │ └── sqlite
│ │ ├── database.py
│ │ └── todo
│ │ ├── todo_repository.py
│ │ └── todo_dto.py
│ ├── presentation
│ │ └── api
│ │ └── todo
│ │ ├── handlers
│ │ │ └── todo_api_route_handler.py
│ │ ├── schemas
│ │ │ └── todo_schema.py
│ │ └── error_messages
│ │ └── todo_error_message.py
│ └── usecase
│ └── todo
│ ├── create_todo_usecase.py
│ ├── update_todo_usecase.py
│ ├── start_todo_usecase.py
│ ├── find_todos_usecase.py
│ ├── find_todo_by_id_usecase.py
│ ├── complete_todo_usecase.py
│ └── delete_todo_usecase.py
└── tests
ドメイン層には、コアとなるビジネスロジックとルールが含まれています。主に以下の要素で構成されています：
- エンティティ
- 値オブジェクト
- リポジトリインターフェース
このプロジェクトでの各コンポーネントの実装は以下の通りです：
エンティティは一意の識別子を持つドメインモデルです。このプロジェクトでは、Todoクラスがエンティティとして実装されています：
class Todo:
def __init__(
self,
id: TodoId,
title: TodoTitle,
description: Optional[TodoDescription] = None,
status: TodoStatus = TodoStatus.NOT_STARTED,
created_at: datetime = datetime.now(),
updated_at: datetime = datetime.now(),
completed_at: Optional[datetime] = None,
):
self._id = id
self._title = title
self._description = description
self._status = status
self._created_at = created_at
self._updated_at = updated_at
self._completed_at = completed_atエンティティの主な特徴：
- 一意の識別子（
id）を持つ - 状態を変更できる（例:
update_title,update_description,start,completeメソッド） - 識別子によって同一性が決定される（
__eq__メソッドの実装） - ファクトリメソッド（例:
create）を通じて生成されることがある
このプロジェクトでは、インスタンスの同一性を id のみによって判断するため、__eq__ メソッドを以下のように実装しています。
def __eq__(self, obj: object) -> bool:
if isinstance(obj, Todo):
return self.id == obj.id
return Falseこの実装のポイント：
- 同一性は識別子（
id）のみによって判断される isinstanceチェックによる型安全性が確保されている
値オブジェクトは識別子を持たない不変のドメインモデルです。このプロジェクトでは、以下のような値オブジェクトを実装しています：
@dataclass(frozen=True)
class TodoTitle:
value: str
def __post_init__(self):
if not self.value:
raise ValueError('Title is required')
if len(self.value) > 100:
raise ValueError('Title must be 100 characters or less')値オブジェクトの主な特徴：
@dataclass(frozen=True)による不変性の保証- 値の検証ロジックを含む（
__post_init__） - 識別子を持たない
- すべての値の内容によって同一性が決定される
リポジトリはエンティティの永続化を担当する抽象化レイヤーです。このプロジェクトではTodoRepositoryインターフェースを次のように定義しています：
class TodoRepository(ABC):
@abstractmethod
def save(self, todo: Todo) -> None:
"""Save a Todo"""
@abstractmethod
def find_by_id(self, todo_id: TodoId) -> Optional[Todo]:
"""Find a Todo by ID"""
@abstractmethod
def find_all(self) -> List[Todo]:
"""Get all Todos"""
@abstractmethod
def delete(self, todo_id: TodoId) -> None:
"""Delete a Todo by ID"""リポジトリの主な特徴：
- エンティティの永続化を抽象化する
インフラ層はドメイン層で定義されたインターフェースを実装します。主に以下の要素で構成されています：
- データベース設定
- リポジトリの実装
- 外部サービスとの統合
- 依存性注入（DI）の設定
この層は、ドメイン層で定義された抽象化の具象実装を提供し、関心の明確な分離を維持しながら、アプリケーションが外部リソースと相互作用できるようにします。
リポジトリの実装は、次のように行います：
class TodoRepositoryImpl(TodoRepository):
"""SQLite implementation of Todo repository interface."""
def __init__(self, session: Session):
"""Initialize repository with SQLAlchemy session."""
self.session = session
def find_by_id(self, todo_id: TodoId) -> Optional[Todo]:
"""Find a Todo by its ID."""
try:
row = self.session.query(TodoDTO).filter_by(id=todo_id.value).one()
except NoResultFound:
return None
return row.to_entity()
def save(self, todo: Todo) -> None:
"""Save a new Todo item."""
todo_dto = TodoDTO.from_entity(todo)
try:
existing_todo = (
self.session.query(TodoDTO).filter_by(id=todo.id.value).one()
)
except NoResultFound:
self.session.add(todo_dto)
else:
existing_todo.title = todo_dto.title
existing_todo.description = todo_dto.description
existing_todo.status = todo_dto.status
existing_todo.updated_at = todo_dto.updated_at
existing_todo.completed_at = todo_dto.completed_atリポジトリインターフェースとは異なり、インフラ層の実装コードには、特定の技術（この例ではSQLite）に依存する詳細が含まれていても問題ありません。むしろ、抽象的なインターフェース定義にとらわれすぎず、具体的な技術名をディレクトリ名（例: sqlite）やクラス名に含めることで、その実装がどの技術に基づいているかを明確にすることが推奨されます。
オニオンアーキテクチャでは、内側のレイヤー（ドメイン層）は外側のレイヤー（インフラ層、プレゼンテーション層）に依存しません。そのため、レイヤー間でデータをやり取りする際に、特定のレイヤーの詳細（例えば、インフラ層のデータベースモデル）が他のレイヤーに漏れ出ないように、オブジェクトの変換が必要になることがあります。この変換の役割を担うのがData Transfer Object（DTO）です。DTOは、レイヤー間でデータを転送するために使用されるシンプルなオブジェクトです。
TodoDTO クラスの例を以下に示します。これは SQLAlchemy のモデル（Base を継承）であり、ドメインエンティティ (Todo) との間で相互変換を行うメソッド (to_entity, from_entity) を持ちます。
class TodoDTO(Base):
"""Data Transfer Object for Todo entity in SQLite database."""
__tablename__ = 'todo'
id: Mapped[UUID] = mapped_column(primary_key=True, autoincrement=False)
title: Mapped[str] = mapped_column(String(100), nullable=False)
description: Mapped[str] = mapped_column(String(1000), nullable=True)
status: Mapped[str] = mapped_column(index=True, nullable=False)
created_at: Mapped[int] = mapped_column(index=True, nullable=False)
updated_at: Mapped[int] = mapped_column(index=True, nullable=False)
completed_at: Mapped[int] = mapped_column(index=True, nullable=True)
def to_entity(self) -> Todo:
"""Convert DTO to domain entity."""
return Todo(
TodoId(self.id),
TodoTitle(self.title),
TodoDescription(self.description),
TodoStatus(self.status),
datetime.fromtimestamp(self.created_at / 1000, tz=timezone.utc),
datetime.fromtimestamp(self.updated_at / 1000, tz=timezone.utc),
datetime.fromtimestamp(self.completed_at / 1000, tz=timezone.utc)
if self.completed_at
else None,
)
@staticmethod
def from_entity(todo: Todo) -> 'TodoDTO':
"""Convert domain entity to DTO."""
return TodoDTO(
id=todo.id.value,
title=todo.title.value,
description=todo.description.value if todo.description else None,
status=todo.status.value,
created_at=int(todo.created_at.timestamp() * 1000),
updated_at=int(todo.updated_at.timestamp() * 1000),
completed_at=int(todo.completed_at.timestamp() * 1000)
if todo.completed_at
else None,
)データベースから取得した TodoDTO オブジェクト（SQLAlchemyに依存）を、ドメイン層の Todo エンティティに変換してからユースケース層に返すことで、ユースケース層がインフラストラクチャ層の詳細に依存することを防ぎます。これにより、リポジトリインターフェースで定義された戻り値の型（Todo エンティティ）との整合性も保たれます。
このプロジェクトでは、FastAPIの依存性注入システムを使用してレイヤー間の依存関係を管理しています。DI設定は infrastructure/di/injection.py モジュールに集約されています：
def get_session() -> Iterator[Session]:
"""リクエスト処理用の管理されたSQLAlchemyセッションを提供します。"""
session: Session = SessionLocal()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
def get_todo_repository(session: Session = Depends(get_session)) -> TodoRepository:
"""現在のセッションに紐づくリポジトリインスタンスを提供します。"""
return new_todo_repository(session)
def get_create_todo_usecase(
todo_repository: TodoRepository = Depends(get_todo_repository),
) -> CreateTodoUseCase:
"""リポジトリが注入されたcreate-todoユースケースを提供します。"""
return new_create_todo_usecase(todo_repository)このアプローチの主な利点：
- ライフサイクル管理: データベースセッションが自動的に管理される（commit/rollback/close）
- テスタビリティ: 依存関係をテストで簡単にモック化または置き換えることができる
- 疎結合: プレゼンテーション層は実装ではなくユースケースインターフェースにのみ依存する
- 単一責任: 各依存性プロバイダーは明確な単一の目的を持つ
ユースケース層には、アプリケーション固有のビジネスルールが含まれています。主に以下の要素で構成されています：
- ユースケースの実装
- ユースケースに関係するエラーハンドリング
このプロジェクトでは、「1つのユースケースに1つのパブリックメソッド」というルールを採用し、各ユースケースを単一のexecuteメソッドを持つ独立したクラスとして実装しています。実装例は以下の通りです：
各ユースケースは以下の構造に従います：
class CreateTodoUseCase:
"""CreateTodoUseCase defines a use case interface for creating a new Todo."""
@abstractmethod
def execute(
self, title: TodoTitle, description: Optional[TodoDescription] = None
) -> Todo:
"""execute creates a new Todo."""
class CreateTodoUseCaseImpl(CreateTodoUseCase):
"""CreateTodoUseCaseImpl implements the use case for creating a new Todo."""
def __init__(self, todo_repository: TodoRepository):
self.todo_repository = todo_repository
def execute(
self, title: TodoTitle, description: Optional[TodoDescription] = None
) -> Todo:
"""execute creates a new Todo."""
todo = Todo.create(title=title, description=description)
self.todo_repository.save(todo)
return todoユースケースの主な特徴：
- ユースケースごとに1つのクラスを用意
- 単一責任の原則に従う設計
ユースケースはドメイン固有のエラーを処理します：
class StartTodoUseCaseImpl(StartTodoUseCase):
# ... __init__ ...
def execute(self, todo_id: TodoId) -> Todo:
todo = self.todo_repository.find_by_id(todo_id)
if todo is None:
raise TodoNotFoundError
if todo.is_completed:
raise TodoAlreadyCompletedError
if todo.status == TodoStatus.IN_PROGRESS:
raise TodoAlreadyStartedError
todo.start()
self.todo_repository.save(todo)
return todoプレゼンテーション層はHTTPリクエストとレスポンスを処理します。主に以下の要素で構成されています：
- FastAPIルートハンドラ
- リクエスト/レスポンスモデル
- 入力検証ロジック
ハンドラはpresentation/apiディレクトリに配置され、アプリケーションのAPI層を形成します。各ドメイン（例：todo）は独自のハンドラ、スキーマ、エラーメッセージ定義を持っています。
- Python 3.13以上とuvがインストールされていることを確認
- このリポジトリをクローン
make installを実行して依存関係をインストールmake devを実行して開発サーバーを起動- APIドキュメントにアクセス：http://127.0.0.1:8000/docs
- VSCodeでこのリポジトリをクローンして開く
- Dev Containers拡張機能をインストール
- コマンドパレット（Cmd/Ctrl+Shift+P）を開き「Dev Containers: Reopen in Container」を選択
- コンテナのビルドが完了したら、ターミナルで
make devを実行 - APIドキュメントにアクセス：http://127.0.0.1:8000/docs
- 新しいTodoを作成する：
curl --location --request POST 'localhost:8000/todos' \
--header 'Content-Type: application/json' \
--data-raw '{
"title": "Implement DDD architecture",
"description": "Create a sample application using DDD principles"
}'- POSTリクエストのレスポンス：
{
"id": "550e8400-e29b-41d4-a716-446655440000",
"title": "Implement DDD architecture",
"description": "Create a sample application using DDD principles",
"status": "not_started",
"created_at": 1614007224642,
"updated_at": 1614007224642
}- Todoを取得する：
curl --location --request GET 'localhost:8000/todos'- GETリクエストのレスポンス：
[
{
"id": "550e8400-e29b-41d4-a716-446655440000",
"title": "Implement DDD architecture",
"description": "Create a sample application using DDD principles",
"status": "not_started",
"created_at": 1614006055213,
"updated_at": 1614006055213
}
]- Todoを開始する：
curl --location --request PATCH 'localhost:8000/todos/550e8400-e29b-41d4-a716-446655440000/start'- Todoを完了する：
curl --location --request PATCH 'localhost:8000/todos/550e8400-e29b-41d4-a716-446655440000/complete'- Todoを更新する：
curl --location --request PUT 'localhost:8000/todos/550e8400-e29b-41d4-a716-446655440000' \
--header 'Content-Type: application/json' \
--data-raw '{
"title": "更新されたタイトル",
"description": "更新された説明"
}'make testこのコマンドは、Pyreflyによる型チェックとpytestによるユニットテストの両方を実行します。
このプロジェクトでは、コード品質を維持するために以下のツールを使用しています：
ruffを使用してコードをフォーマットします：
make formatこのプロジェクトには、Dockerベースの開発環境用の.devcontainer設定が含まれています。これにより、異なるマシン間で一貫した開発環境を確保できます。セットアップ手順については起動方法を参照してください。
このプロジェクトはMITライセンスのもとで公開されています。詳細はLICENSEファイルを参照してください。