🐍

Python + FastAPIでオニオンアーキテクチャ（簡易版）

2023/12/11に公開

Python

FastAPI

オニオンアーキテクチャ

tech

概要

Python、FastAPIを用いてオニオンアーキテクチャ（簡易版）を実現するための解説記事です。

対象読者

PythonやFastAPIを触ったことがある人
オニオンアーキテクチャなどのソフトウェアアーキテクチャに興味がある人

筆者のバックグラウンド

普段はTypeScript（Next.jsを主に利用）を用いたフロントエンド開発者です。

以前はバックエンドエンジニアで主にGoを使ってAWS上でAPIの開発などを行なっていました。

最近はLLMを用いたアプリケーション開発に関わっています。

PythonやFastAPIの経験は浅いです。

この記事を書こうと思った動機

Python、FastAPIでLLMのメッセージを生成するためのAPIを開発していましたが、コードベースが肥大化して修正に手間がかかるようになってしまいました。

そこで過去にGoでやったときに開発体験が良かった、オニオンアーキテクチャをやってみようと思ったことがきっかけです。

この記事で説明すること

Python、FastAPIを用いて簡易的なオニオンアーキテクチャを実現するための方法やディレクトリ構成の解説を行ないます。

この記事で説明しないこと

この記事に出てくるアプリケーションはPythonのOpenAIライブラリやFastAPIを利用していますが、それらの詳しい解説はこの記事では割愛させていだきます。

オニオンアーキテクチャの簡単な解説

ソフトウェアアーキテクチャの1つで依存関係を以下の図に従って整理するアーキテクチャになります。

Onion Architecture

外側の層は内側の層に依存しますが、内側の層は外側の層に依存することはありません。

これにより依存関係が明確になり、コードの変更に強くなり、テストコードも書きやすくなります。

最も重要なドメイン層がフレームワークなどの外部ライブラリに依存していないので、バージョンアップによってビジネスロジックが壊れるといった事も起こりにくいです。

同じようなアーキテクチャで一番最初に知ったのは以下の図で有名なクリーンアーキテクチャだったのですが、クリーンアーキテクチャは実装難易度が高いと感じました。

Clean Architecture

それに比べてオニオンアーキテクチャはシンプルでわかりやすく初心者でもわかりやすいと感じます。
（ちなみにオニオンアーキテクチャもクリーンアーキテクチャも解決しようとしている問題は同じです）

以下の松岡さんの説明が今まで見た中で一番わかりやすかったので今回の実装も以下を参考にしております。

https://youtu.be/80NeuPXs2J0

各レイヤーの実装紹介

サンプルアプリケーション

サンプルとして個人で開発中のねこの人格を持ったAIと会話できるサービスを例に説明します。（このサービスはフロント側はまだ未完成ですが、バックエンド側は最低限の機能が完成しています）

https://www.ai-meow-cat.com/

上記サービスのバックエンド側のソースコードは以下に公開しています。

こちらのコードを見て概要を理解できる人は以降の章を読む必要はありません。

https://github.com/nekochans/ai-cat-api

アプリケーションの実行環境やバージョン情報は下記の通りです。

Python: 3.11.3
FastAPI: 0.101.0
アプリケーション実行環境: Fly.io
データベース: PlanetScale
LLM: OpenAI APIをPythonの openai ライブラリを経由して利用

ディレクトリ構成

ディレクトリ構成は以下の通りです。

このあとで各ディレクトリについて説明していきます。

├── src
│   ├── domain         # ドメイン層
│   ├── log            # ロギング（後で説明します）
│   ├── usecase        # ユースケース層
│   ├── presentation   # プレゼンテーション層
│   ├── infrastructure # インフラストラクチャ層
│   └── main.py        # エントリーポイント

ドメイン層 (`src/domain/`)

依存関係の最も内側の層でビジネスロジックを定義します。

本格的にドメイン駆動設計をやっている場合、Entities（エンティティ）パターンやValue Objectパターンを使って、EntityやValue Objectクラスを実装するかと思いますが、今回は簡易版なのでそこまではやっていません。

ですがビジネスロジック上重要なLLMに渡すプロンプトを生成する関数などを実装しています。

またRepositoryというデザインパターンを使ってLLMやRDBとのやり取りを抽象化するためのインターフェースを定義しています。

Pythonはインターフェースの構文が存在しないので Protocol を使って実装しています。

厳密に言うとドメイン層では外部ライブラリへの依存する際はインターフェースを通して直接依存しないようにするのが原則となります。
しかし uuid や re などのPythonの標準パッケージに関しては依存してしまったほうが実装が簡単なので標準パッケージには依存する形で実装しています。

外部ライブラリは変更が激しいので外部ライブラリへの依存はしないようにしています。

ユースケース層 (`src/usecase/`)

アプリケーションサービス層という名前が正式名称ですが、松岡さんの記事を参考にユースケース層という名前に置き換えています。

アプリケーションサービス層という名前は責務がわかりにくいので、アプリケーションのユースケースを実現するためのユースケース層と読んだほうがわかりやすいという主張には同意です。

ユースケース層はその名の通り、アプリケーションの各ユースケースを実現するための層です。

ドメイン層の関数やクラスを利用して必要なユースケースを実現することが責務となります。

このアプリケーションはWebAPIを実装しているので1つのAPIエンドポイントにつき、1つのユースケースクラスが実装される形になります。

以下はその実装例です。こちらは未ログインのユーザー向けに提供している会話機能を実現しています。

DTO（Data Transfer Object）を引数として受け取って TypedDict で定義されたとおりのPythonの辞書を返します。

この層はFastAPIたDB操作用の外部ライブラリに依存しません。よってDTOのなかでRepositoryクラスのインターフェースを受け取るようにすることで依存を回避しています。

ちなみにトランザクション制御もここで行なっています。

外部のDBライブラリに直接依存しないように DbHandlerInterface を同階層に定義して、これをDTOで受け取ることで依存を回避しています。

ドメイン層、ユースケース層についての補足

ドメイン層とユースケース層は外部ライブラリやフレームワークに依存していない層になります。

その為、この層のテストコードをちゃんと書いておけばフレームワークのバージョンアップ時の動作確認が楽になります。

特にユースケース層のテストコードは対象のユースケースのカバレッジをカバーできるので、テストコードはユースケース層のテストを多めに書くのが基本となります。

プレゼンテーション層 (`src/presentation/`)

アプリケーション外部との入出力を実現するための層です。

このアプリケーションはWebAPIなのでHTTPのリクエストオブジェクトやルーティングを定義して最終的にJSONやServer Sent Events(SSE)形式のレスポンスを返すことがこの層の役割になります。

その特性上フレームワークの機能に大きく依存します。

このアプリケーションだとFastAPIやpydanticなどのパッケージの機能を多く利用しています。

ルーティング（HTTPのルーティングの役割を果たす、FastAPIの機能を利用）

Controller（ユースケース層の機能を利用して最終的にHTTPレスポンスを生成する責務を担う）

ちなみにControllerから直接インフラストラクチャ層（あとで説明）のDB操作クラスやRepositoryの実装クラスを直接利用していますが、これは本来のオニオンアーキテクチャではアーキテクチャ違反になります。

インフラストラクチャ層のクラスもインターフェースを通じて依存を回避するのが正しい実装ですが、プレゼンテーション層はフレームワークへ依存してしまったほうが実装がシンプルになるのでここは許容しています。

その他、APIの認証に関してもこの層で実装されています。

インフラストラクチャ層(`src/infrastructure/`)

技術的な機能を実現するための層です。

この層は外部ライブラリ等への依存が多くなります。

トークンの消費量を確認するための tiktoken のようなライブラリを使った処理やデータベースに接続するためのクラスが実装されています。

またRepositoryクラスの実装クラスもここで実装されています。

例えば以下は GuestUsersConversationHistoryRepositoryInterface の実装クラスです。

利用しているライブラリが aiomysql なので src/infrastructure/repository/aiomysql/ に実装されています。

今後別のライブラリ（例えばSQLAlchemy等）での実装を行なう場合は src/infrastructure/repository/sqlalchemy/ が作られる事になります。

GuestUsersConversationHistoryRepositoryInterface のインターフェースに従うことでRepositoryクラスを利用する側は使い方が変わらないことが保証されます。

以下はAIのメッセージを生成するためのRepositoryクラスです。

GuestUsersConversationHistoryRepositoryInterface と同じように CatMessageRepositoryInterface のインターフェースに準拠するように実装されています。

こちらも openai のライブラリを使っているので src/infrastructure/repository/openai/ に実装されています。

こちらも今後 LangChain などのライブラリを使って実装される場合は src/infrastructure/repository/langchain/ 配下に実装される事になります。

その他

エントリーポイント(`src/main.py`)

HTTPリクエストが送信された際に最初に呼ばれるファイルになります。

アプリケーションサーバーの起動とルーティングの読み込みだけにしたかったのですが、404エラーなどのレスポンスを整形するための @app.exception_handler が定義されています。

ロギング（`src/log`）

ログを出力するための機能です。

ログ出力はどの層でも実行する可能性があるのでどこに置くか迷いました。

ドメイン層に置けば、それ以外の層で使うことは可能ですがロギングの機能がドメイン層にあるのは違和感があります。

最終的に以下の投稿を参考にどこにも属さないロギング用の層を作ることで解決しました。

ちなみにロギングの実装は以下のようになっています。

以下のようなJSON形式のログを出力しています。

成功ログの例

{
	"name": "root",
	"msg": "success",
	"args": [],
	"levelname": "INFO",
	"levelno": 20,
	"pathname": "/src/usecase/generate_cat_message_for_guest_user_use_case.py",
	"filename": "generate_cat_message_for_guest_user_use_case.py",
	"module": "generate_cat_message_for_guest_user_use_case",
	"exc_text": null,
	"stack_info": null,
	"lineno": 168,
	"funcName": "execute",
	"created": 1702190986.977943,
	"msecs": 977,
	"relativeCreated": 113538.92016410828,
	"thread": 281473372679488,
	"threadName": "MainThread",
	"processName": "SpawnProcess-1",
	"process": 8,
	"request_id": "572964f7-f6a7-440f-a41b-6703e705b5ad",
	"conversation_id": "572964f7-f6a7-440f-a41b-6703e705b5ad",
	"cat_id": "moko",
	"user_id": "6a17f37c-996e-7782-fefd-d71eb7eaaa37",
	"ai_response_id": "chatcmpl-00000000000000000000000000000"
}

エラーログの例

{
	"name": "root",
	"msg": "An error occurred while connecting to the database: (1054, 'target: ai_cat_test.-.primary: vttablet: rpc error: code = NotFound desc = Unknown column \\'ai_message__\\' in \\'field list\\' (errno 1054) (sqlstate 42S22) (CallerID: esgr25wn9pkxrcov0z1k): Sql: \"select user_message, ai_message__ from guest_users_conversation_histories where conversation_id = :conversation_id /* VARCHAR */ order by id desc limit :vtg1 /* INT64 */\", BindVars: {REDACTED}')",
	"args": [],
	"levelname": "ERROR",
	"levelno": 40,
	"pathname": "/src/usecase/generate_cat_message_for_guest_user_use_case.py",
	"filename": "generate_cat_message_for_guest_user_use_case.py",
	"module": "generate_cat_message_for_guest_user_use_case",
	"exc_text": null,
	"stack_info": null,
	"lineno": 99,
	"funcName": "execute",
	"created": 1702191464.4318147,
	"msecs": 431,
	"relativeCreated": 4497.754096984863,
	"thread": 281473389890880,
	"threadName": "MainThread",
	"processName": "SpawnProcess-2",
	"process": 189,
	"request_id": "284cd7e6-2bec-4013-81a2-69ea7d494a3e",
	"conversation_id": "284cd7e6-2bec-4013-81a2-69ea7d494a3e",
	"cat_id": "moko",
	"user_id": "6a17f37c-996e-7782-fefd-d71eb7eaaa37",
	"user_message": "こんにちはもこちゃん🐱",
	"traceback": [
		"Traceback (most recent call last):",
		"  File \"/src/usecase/generate_cat_message_for_guest_user_use_case.py\", line 89, in execute",
		"    chat_messages = await self.dto[",
		"                    ^^^^^^^^^^^^^^^",
		"  File \"/src/infrastructure/repository/aiomysql/aiomysql_guest_users_conversation_history_repository.py\", line 30, in create_messages_with_conversation_history",
		"    await cursor.execute(sql, (dto[\"conversation_id\"],))",
		"  File \"/usr/local/lib/python3.11/site-packages/aiomysql/cursors.py\", line 239, in execute",
		"    await self._query(query)",
		"  File \"/usr/local/lib/python3.11/site-packages/aiomysql/cursors.py\", line 457, in _query",
		"    await conn.query(q)",
		"  File \"/usr/local/lib/python3.11/site-packages/aiomysql/connection.py\", line 469, in query",
		"    await self._read_query_result(unbuffered=unbuffered)",
		"  File \"/usr/local/lib/python3.11/site-packages/aiomysql/connection.py\", line 683, in _read_query_result",
		"    await result.read()",
		"  File \"/usr/local/lib/python3.11/site-packages/aiomysql/connection.py\", line 1164, in read",
		"    first_packet = await self.connection._read_packet()",
		"                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^",
		"  File \"/usr/local/lib/python3.11/site-packages/aiomysql/connection.py\", line 652, in _read_packet",
		"    packet.raise_for_error()",
		"  File \"/usr/local/lib/python3.11/site-packages/pymysql/protocol.py\", line 221, in raise_for_error",
		"    err.raise_mysql_exception(self._data)",
		"  File \"/usr/local/lib/python3.11/site-packages/pymysql/err.py\", line 143, in raise_mysql_exception",
		"    raise errorclass(errno, errval)",
		"pymysql.err.OperationalError: (1054, 'target: ai_cat_test.-.primary: vttablet: rpc error: code = NotFound desc = Unknown column \\'ai_message__\\' in \\'field list\\' (errno 1054) (sqlstate 42S22) (CallerID: esgr25wn9pkxrcov0z1k): Sql: \"select user_message, ai_message__ from guest_users_conversation_histories where conversation_id = :conversation_id /* VARCHAR */ order by id desc limit :vtg1 /* INT64 */\", BindVars: {REDACTED}')"
	]
}

テストコードについて

tests/ 配下にテストコードを書いています。

さきほど記載したとおり、基本的にはユースケース層のテストを中心に記載しています。

ここにしっかりテストを書けば、一番重要なドメイン層のコードカバレッジもある程度増えるし、ドメイン層のリファクタリングなどを実施したときも機能が壊れていないことが保証しやすいからです。

以下はユースケース層のテストです。

ユースケース層のテストだとDBやLLMの応答を生成する部分はMockを利用しています。

Repositoryクラスを丸ごとMockに置き換えている形です。

ただしRepositoryの実装クラスのテストがないと、DBの操作のテストなどができないのでRepositoryクラスのテストは実装しています。

DBのテストはテスト用のDBを利用して実際のDBに繋いでいます。

ここをMockで終わらせてしまうとDBの操作部分がテストされないことになるのでテストの実行速度が多少落ちても実際のDBを使ってテストしています。

以下の記事でDBを使ったテストを高速化するための手法を紹介しているので興味がある人は見てくれると嬉しいです。

型チェック

テストコードではないですが mypy を使った型チェックを行なっています。

実装コストが多少大きくなる代わりにバグを防いでくれたり、コードの読みやすさやIDEと組み合わせたときの生産性の向上などのメリットのほうが大きいので基本的には何らかの仕組みで型チェックを実行できるようにしたほうが良いと考えています。

今後は mypy ではなく別のツールを利用する可能性もあります。

おわりに

簡単ですが以上がPython + FastAPIでオニオンアーキテクチャ（簡易版）の解説でした。

PythonもFastAPIも慣れていなかったのですが、コードベースが肥大化やテストが書きにくい問題は解決できたのでやって良かったと思っています。

以上になります。最後まで読んでいただきありがとうございました。