アミフィアブル株式会社AIチームのエンジニアの野村です。
はじめに
私が所属するPython/AIチームでは、PythonによるREST APIの開発を行っています。その開発には、静的型チェックツールのmypyとlintツールruffの使用が前提となっており、Pythonコードに厳密な型付けが義務付けられています。（以降のPythonコードは型付けが前提となります）。
厳密に型付けを行うことで本来であれば型無しの動的オブジェクト指向言語という関数型言語とは対照的なPythonであっても関数型のライブラリを導入しやすくなります。そこで、私のチームではPythonの関数型プログラミングライブラリreturns(公式サイト)が提供するResultモナドを導入しています。
returnsのResultは関数型言語のResultモナドないしはEitherモナドを模倣して実装されていますが、returnsのドキュメントでは、モナドという単語は使用せずコンテナ(container)という言い回しを使っています。また、関数型ドメイン設計や鉄道指向プログラミング解説でもモナドという用語は避けられてResult型と呼ばれることが多いようです。実際のところ関数に副作用を含みうるPythonでは定義上モナドとは言えません。この点に関しては別の記事で解説したいと思います。ただ、この記事ではモナドを恐れず「モナド」という用語で統一します。
Pythonでモナドといったガチガチの関数型スタイルを導入するメリットはなんでしょうか。結論を先に言いますと、Resultモナド導入は次の2つが大きな動機になります：
- オニオンアーキテクチャのレイヤー構造との親和性
- エラーハンドリングの厳密化と明晰化
これらの主題の前に下ごしらえとしてモナドを簡単に説明します（Part1）。また、チーム開発でのResult導入においてdo記法の役割が大きい点にも触れます（Part2）。そして、モナドを扱う上で必要不可欠となりまたオニオンアーキテクチャでのレイヤー区別に関係してくる純粋関数／不純な関数の区別を説明します（Part3）。その後、Resultを導入したオニオンアーキテクチャ（Part4)とエラーハンドリング（Part5）を解説します。
なお、対象読者としては、関数型プログラミング、および、関数型ドメインモデリングに関心があるPythonエンジニアを想定しています。
採用するオニオンアーキテクチャ
私が所属するAIチームでは、microsoftのオニオンアーキテクチャを参考にしたアーキテクチャを採用しています。フォルダ構成や名称なども基本的にこれを踏襲しておりますが、ただ採用しているフレームワークがflaskではなくFastAPIであり、データバリデーションライブラリもMarshmallowではなくPydanticを採用しています。これらの違いはアーキテクチャ設計に大きな違いはもたらしません。
レイヤーの概要
この記事ではオニオンアーキテクチャの解説は行いません。多くの解説が見つかりますし、Microsoftのレポジトリを落として自分で観察してみるのが一番わかり易いと思います。ただ、各レイヤーの名称はいろいろな紹介文でブレがありますので、上述のサンプルより主要レイヤーだけ切り出し概説をは与えるとともに名称を固定しておきます。
├── src
│ ├── api
│ │ └── controllers
│ │ └── ... # controllers for the api
│ ├── services
│ │ └── ... # Services for interacting with the domain (business logic)
│ ├── infrastructure
│ │ └── services
│ └── domain
│ └── ... # Business logic models
1. ドメイン層 (domain)
純粋なビジネスロジックを含みます。データベースや外部API、ファイルの入出力への依存は一切ありません。
2. インフラ層 (infrastructure)
外部と連携を行う層です。データベースやAPI、ファイルシステムなどの外部サービスと接続する責任を負います。
3. サービス層 (services)
ドメインエンティティを協調させることでユースケース（Use Cases）を実装します。
4. API層 (api)
アプリケーションをHTTPなどを通して外部に公開する最も外側のレイヤーです。この層ではリクエスト/レスポンス処理、シリアライズ、バリデーション、エラーレスポンスの整形などを行います。FastAPIを通してOpenAPIを作成するのもこの層で行います。
各層の区別
さて、オニオンアーキテクチャの各層をこれまで準備してきたResultモナドと純粋/不純関数といった関数型のスタイルで区別していきたいと思います。また、各レイヤーでResultとIOResultを使い分けることでそれぞれの特徴を明確にします。pythonに関数型のスタイルを導入したのは、Resultモナドで厳密かつ明晰なエラーハンドリングを行えるという点だけではなく、オニオンの層の区別がより明晰になるからです。つまり、
- ドメイン層は可能な限り純粋関数で構成します
- インフラストラクチャ層は不純な関数で構成し副作用（外部連携する機能）をここに集めます
- サービス層は上記ふたつをdo記法で結びつけてサービス/ユースケースを描写します
- API層はそのようにまとめられたユースケースをAPIのI/Fを提供します
このような特徴づけを行うことで、各層で行う責務が明晰かつ厳格に区別することができます。ここで大事なのは、ドメイン層を純粋関数で構成し、不純な処理は境界（インフラ層）に隔離する、ということです。関数の特徴でオニオンアーキテクチャといった層の特徴づけを行うことでオニオンアーキテクチャの初心者であっても、どの関数をどの層に配置したらいいのかという判断の手助けになります。
ただ注意しなければならないのは、純粋/不純だけで層の区別するのは間違っています。不純な関数がドメイン層に含まれることはありませんが、純粋関数が上位レイヤーに含まれることはあるからです。例えば単純なヘルパー関数などは純粋関数になるかもしれませんが、置き場所は必ずしもドメイン層とはならないでしょう。そのため、純粋と不純の区別は、あくまで階層構造と役割分担の理解の手助けとしてこの区別は機能します。
Resultモナドによる層の分類
また各層を関数の純粋さで区別するということは、Resultモナドの区別を導入できます。つまり、returnsには、Resultモナドは二種類あり、ResultとIOResultです。前者はいままで説明してきたモナドで純粋関数をラップしてエラーという副作用を隔離するために使います後者はIOの成功失敗を扱うモナドです。具体例を見たほうが理解が早いと思いますので、簡単な業務ロジックを組み上げていきます。任意のstring（例えば、taro）をリクエストとして投げたら、hello mr. taroと挨拶加えたテキストファイルをレスポンスとして返してくれる挨拶APIを考えていきましょう。
ドメイン層（doamin）
from __future__ import annotations
from fastapi import HTTPException
# NOTE: domain層ではResultを使う。IOResultはinfrastructure層より上の階層で使い、純粋関数と不純関数の出力を区別する。
from returns.result import Failure, Result, Success
def is_empty_string(x: str) -> bool:
return x == ""
# NOTE: 関数はservice層で純粋関数と不純関数をbindでチェーンさせるために、hoge(input)->Result[output, error_code]という関数のシグネチャで統一する。引数にResultを使わないようにする。
def add_hello(xstr: str) -> Result[str, DomValueError]:
if is_empty_string(xstr):
return Failure(ValueError())
return Success("hello " + xstr)
def add_mr(xstr: str | None) -> Result[str, DomValueError]:
if xstr is None:
return Success("mr. nanashi")
if is_empty_string(xstr):
return Failure(DomValueError())
return Success("mr." + xstr)
インフラ層（infrastructure）
from openpyxl import Workbook
# インフラ層では不純な関数、特に外部との入出力を扱うのでIOResultを使う。
from returns.io import IOFailure, IOResult, IOSuccess
class TextWriter:
@staticmethod
def write_name(name: str) -> IOResult[str, str]:
try:
text_content = f"名前: {name}\n"
return IOSuccess(text_content)
except Exception as e:
return IOFailure(str(e))
@staticmethod
def streaming_response(text_content: str) -> IOResult[StreamingResponse, str]:
try:
filename = "skeleton_sample.txt"
output_stream = io.BytesIO(text_content.encode("utf-8"))
output_stream.seek(0)
response = StreamingResponse(
output_stream,
headers={"Content-Disposition": f"attachment; filename={filename}"},
media_type="text/plain",
)
return IOSuccess(response)
except Exception as e:
return IOFailure(str(e))
サービス層（services）
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from fastapi import HTTPException
from fastapi.responses import StreamingResponse
from selenium.common.exceptions import NoSuchElementException, WebDriverException
from returns.io import IOResult
from api_skeleton.domain_layer import sample_code_str
from api_skeleton.infrastructure_layer.excel_reader_writer import ExcelReaderWriter
from api_skeleton.infrastructure_layer.html_reader import HtmlReader
from api_skeleton.infrastructure_layer.print_console import PrintConsole
class GreetingService:
def __init__(self, text: str | None) -> None:
self.text = text
def add_hello_mr(self) -> IOResult[StreamingResponse, HTTPException]:
return IOResult.do(
streaming
# ▼▼▼ api layer ▼▼▼
for checked_string in IOResult.from_value(self.text)
# ▼▼▼ infrastructure layer ▼▼▼
# ▼▼▼ domain layer ▼▼▼
for adding_mr in IOResult.from_result(sample_code_str.add_mr(checked_string)) # resultモナドをIOResultにシフトして統一する
for adding_hello in IOResult.from_result(sample_code_str.add_hello(adding_mr))
# ▲▲▲ domain layer ▲▲▲
for excel_wb in ExcelReaderWriter().write_name(schema)
for streaming in ExcelReaderWriter().steaming_response(excel_wb)
# ▲▲▲ infrastructure layer ▲▲▲
for output in IOResult.from_value(adding_hello)
)
API層（api）
@router.post(
# エンドポイントを定義するAPIの玄関部分でモナドを剥がすのはここだけ
"/hello-mr/excel",
response_class=StreamingResponse,
)
def hello_mr_html_impl(body: HtmlInfoSchema) -> Any:
# エラーハンドリングに失敗しているものはここでキャッチする。
try:
result = GreetingWebsiteService(url=body.url, xpath=body.xpath).add_hello_mr_name_in_website_stream()
except Exception as e:
HTTPException(status_code=500, detail=e)
# Responseを返す最後の最後にモナドから値を取り出す
if is_successful(result):
var: IO[StreamingResponse] = result.unwrap()
return unsafe_perform_io(var)
# エラーも同様に最後にモナドから取り出す
raise unsafe_perform_io(result.failure())
オニオンアーキテクチャにResultモナドを導入するメリット
- テスト観点が明確になる
純粋関数だけでできたロジックは、モックなしでユニットテストが書け、純粋に業務ロジックをテストできる。
不純な関数部分については「外部とのI/Oが絡むテスト」として切り分けられるため、必要に応じて結合テストやシステムテストに集中できる。 - ソースコードの可読性・保守性が向上する
ドメイン層で、「この関数は必ず副作用を起こさない」とわかっていれば、コードレビューやリファクタリングの際に頭の中でシミュレーションしやすい。逆に副作用があればすぐに修正対象と判断できる。
外部リソースとのやりとりに関する処理はインフラ層に閉じ込められるため、潜在的なバグや変更時の影響範囲が特定しやすい。 - アーキテクチャの厳密化
オニオンアーキテクチャでは「ドメインを純粋に保つ」ことが推奨されますが、Python など非純粋な言語ではあいまいになりがち。関数の純粋／不純を常に意識するようにしてResltとIOResltをレイヤーごとに使い分けすると、アーキテクチャのレイヤー分離が自然に担保される。
前回の記事でも触れましたが、Python では言語的に副作用を強制的に分離する機構がないため、エンジニア自身が副作用を把握して排除したり意図的に含める必要があり、これが一番の懸念点となります。しかし、コードレビューや認識共有でその点を徹底することで、Resultモナドをオニオンアーキテクチャに導入する恩恵を受けられるようになります。