(ほぼほぼ)typed (わりかし)fuctional python【関数型ドメインモデリング】

動機

• Domain Modeling Made Functional の内容にインスパイヤされ、型->関数->型的なアプローチの素振りをしたい。

  • あとはこの辺の内容をつまんでまとめることを目指しました。ロバストpython、なっとく！関数型プログラミング

• webバックエンドにおいても、型+関数ベースでモデリングしていく発想への言及をちょくちょく見かける気がするから【第5回】「型」はウェブシステム開発に「エンドゲーム」をもたらすか

  • もしかしかたら、純粋関数、イミュータブル、継承より委譲などの(一般的に合意が得られていると思われる)プラクティスを行うと、class主体ではなく、型->関数->型およびその集合が素直な表現になってくるケースがあるのではないか？
  • 状態はフロントエンドやインメモリDB、もしくはDBが持つことが多く、APIはステートレスにすることが多いのではないか。
  • 単純にデータソース->jsonのパイプライン的な発想でモデリングで十分な場面は多いのではないか。
  • React + TSの流行による関数型(的なもの)+型アプローチの土壌の形成

• pythonを題材としたのは

  • 最近自分が業務で使用しているから。
  • 間口が広いから。(Rustなどのモダン言語を使いたいからと言って即採用できるような現場は、日本のIT現場全体で見れば必ずしも多くないはずです)
  • typing系のdataclass、パターンマッチなどの道具が充実してきていると思うから。

私は、純粋関数型言語に精通しているわけでもなければ、まして難しいアカデミックな概念もよくわからないので、以下のような概念を軽量に、非原理主義的に、普段の開発に取り入れることを目的としたいと思います。

• エラーハンドリングや逐次処理のパイプラインをすっきりさせる(ドメインロジックから分離する)ためResultを導入する
• ドメインに登場するデータや状態を型としてモデリングし、他の（あるいは後の）開発者に向けてよりコード(型)でドメイン/要件を語るようにし、またコンパイルタイムでIDEによる迅速なエラーフィードバックを得たい
• 可能な限りコア部分を純粋にして、テストをストレスなく書きたい。

後に使用するサンプルコード全体のリポジトリ

サンプルストーリー

今回のお題とする、架空のサービスについて説明します。
全てがアナログな方法で実施されている、「商品受注」の非常に単純なワークフローをシステム化していくことを想定します。

注文の全体の流れ

「ウッドパラダイス」という、中規模の家具店の1日を想像してみましょう。(店名はchatGPTのセンスです)

ウッドパラダイスは、町の中心部に位置する家具の老舗です。家具の注文・受注は、すべてアナログ手段で行われています。あなたはウッドパラダイスの社長から、サービスのwebアプリケーション化の依頼を受けたソフトウェアエンジニアです。ここで、あなたは対象のドメイン知識を自分のものとするため、ある日の顧客の業務を直接観察することにしました。

• 顧客の訪問:
  一日の始まり、若いカップルが店に訪れました。店内を見て回り、彼らの目に留まったのは、深緑のヴェルヴェットのソファでした。
  顧客はソファに座ってその快適さを確認し、これに決めたいと窓口に申し込みます。

• フロント業務（店舗窓口）での「注文イベント」発生:
  窓口担当（フロント業務）の山本さんは、特製の注文書に、ヘッダーに「新規注文」と、用紙の中身部分に「注文品、個数、顧客の情報、配送先」を記入し、バックオフィスの受注チームに渡します。
  この用紙は、商品受注のワークフローを起動する「コマンド」として機能しています。

• 商品受注ワークフロー:
  受注チームは、山本さんから渡された注文書の内容を確認し、記載内容の正しさのチェックや、金額計算、到着日の決定を行い、その結果を最終的に発送票付き請求書に転記します。
  その後、注文が完了したことと発送の詳細(あるいは不備があり、失敗したこと)を、フロントに返答します。

• 商品発送チームへの指示:
  受注チームは次に、商品発送チームに注文の詳細を伝えるため、　発送表付き請求書を専用のポストに投函します。
  商品発送チームは、彼らの業務状況に応じたタイミングで順次ポストから封筒（発送のコマンド）を取り出し、指定された日時にソファを顧客の住所まで配送することとなります。
  (支払いは現物代引きと仮定しましょう)

さて、ここで今回は、商品受注ワークフローにフォーカスしてサービスをエンコーディングしていくことを考えましょう。
ワークフローは複数のタスクからなり、以下のステップとして表現できます。

それぞれのタスクの中身や、その他必要な補足をします。

タスクについて

• 簡単のため、ワークフローには３つのステップだけが存在するとします。
  • 注文内容チェック: 住所などの記載ミスがないか、数量に異常はないか確認する
  • 商品金額算出: 商品の現在の価格や注文個数から、商品の合計金額を算出する
  • 到着日決定: 配送先のエリアに応じて、商品到着に要する日数を決定する
• それぞれのタスクは、in-process-outが定義されています。in/outの情報は決まった形を持っています。
  • 注文内容チェック
    • in : 未検証の注文書
    • out: 検証済みの注文書
  • 商品価格計算
    • in : 検証済みの注文書
    • out: 請求書
  • 到着日決定
    • in : 請求書
    • out: 発送票付き請求書

我々が普段仕事でタスクを実行する場合と同じようなものでしょうか。
インプット（や、期限/制約/大目的といったコンテキスト）に付加価値を加え、なんらかのアウトプットを出すという作業を繰り返していると思います。(in-process-out)

もっと言えば、我々がエンコードする対象のドメインはそういった「仕事」であるケースが多いのではないでしょうか。(界隈による、でしかないとは思いますが)

なので今回はそれを、「注文」や、「カリキュレイター」、「注文コントローラー」というのような主体が内部状態とそれを変更する振る舞いを持ち、それらが相互作用していく、というような複雑な捉え方はせず、ただただ素直にin-process-out(IPO)が集まり一つのIPOを形成している、と捉えておくこととします。

閑話休題。サンプルストーリーに話を戻します。

• 各タスクの結果は、指定の封筒に入れて次のタスクの係に渡すことでやりとりする決まりになっています（重要）。

  • 封筒には、タスクが正常に終了した場合の「アウトプット書類」orなんらかの不備があった場合のメモの「不備報告書」、必ずどちらか一方を入れるという決まりがあります。(この封筒は、成功or失敗という文脈を内包している、とも言えるかもしれません。)
    
  • 前工程の係から渡された封筒に、失敗した場合のメモが入っていた場合は、何もせず封筒に入れたまま後工程の係の者に渡すことになっています。（最終的に、失敗メモが窓口に渡され、顧客にその旨を伝えることになるでしょう）
  • またこの封筒の表面には、ワークフローとして順番に実施すべきタスクが、付箋により紐づけ、「バインド」されています。
    
  • 今回の場合は、上から「注文内容チェック」「商品金額算出」「到着日決定」と書いた付箋が接着されていることでしょう。（封筒にタスクがバインドされているとも、タスク同士のout→inがバインドされているとも言えますね）
  • 例えば、注文内容チェック係は、封筒に入った未検証の注文書を受け取り、それを取り出し、OKだったら検証済みの注文書を新規に起票し、それを再度封筒に入れて次の係に渡すことになります。
    (未検証の注文書を破棄して、検証済みの注文書を新しい用紙として作成するイメージを持っていただければと思います。現実だったらエコではないし非効率ですが。)

ストーリーのエンコーディング

今回実装するのは、バックオフィスの業務の商品受注ワークフローの部分です。(つまり商品注文のbacking serviceの実装をする)

• サンプルコードの環境: python3.12(説明にはあまり関係ないがFastAPIで構築)
• 型チェッカ: Pyright(VSCode拡張のPylanceを使用)

考え方

まず、要件を以下の４つの観点のどれかに分類して捉えつつ、実装していこうと思います。

• データモデル -> 型でエンコーディング
• ビジネスロジック^[1] -> 関数でエンコーディング
• ビジネスルール^[2] -> 型あるいはメソッドでエンコーディング
• データアクセス -> 関数でエンコーディング

そしてもちろん

• 共通化

すべき部分は括り出していきます。

要件のまとめ

ここで運よく顧客より、より詳細な商品受注ワークフローの業務マニュアルを入手しました。これを元に観察の結果得た知識を加えて、文書化してみました。
前述の設計観点を踏まえつつ、これをpythonコードに翻訳していきましょう。

商品受注業務マニュアル 全文

商品受注業務マニュアル

1. 業務の全体像

1.1 「商品受注」全体のinputと成果物

• input: フロント業務より受け取った書類「注文書」
• 成果物: 発送票付き請求書 または 不備報告書

1.2 全体の動きの流れ

初期注文受付
-> 注文内容のチェック
-> 価格の計算
-> 配送日の決定

2. 書類とその項目

2.1 初期注文書

• 記載項目: 商品ID、数量、注文者都道府県、注文者市町村区以下住所。

2.2 確認済み注文書

• 記載項目: 商品ID、数量、配送先住所。

2.3 請求書

• 記載項目: 商品ID、数量、合計価格、配送先住所。

2.4 発送票付き請求書

• 記載項目: 合計価格、配送先住所、到着予定日。

2.5 エラー報告書

• 記載項目: エラーコード、エラーメッセージ。

3. 各タスクの手順

3.1 初期注文受付

• フロントから注文書を受け取り、この書類を封筒に入れ、後続の係に渡す。

3.2 注文内容のチェック

• input: 注文書
• 成果物: 検証済み注文書
• 仕事: 封筒から書類を取り出し、住所の有効性と数量の妥当性をチェック。不備がある場合は不備報告書を作成。書類を封筒に入れ、後続の係に渡す。

3.3 価格の計算

• input: 検証済み注文書
• 成果物: 請求書
• 仕事: 封筒から書類を取り出し、商品情報に基づき価格を決定し、請求書を作成。商品情報が不足している場合は不備報告書を作成。書類を封筒に入れ、後続の係に渡す。

3.4 配送日の決定

• input: 請求書
• 成果物: 発送票付き請求書
• 仕事: 封筒から書類を取り出し、配送先に基づき配送日数を計算し、発送票付き請求書 に記載。配送不可能な地域の場合は不備報告書を作成。書類を封筒に入れ、後続の係に渡す。

いざ実装

まず、httpやシステム的なインフラを気にしないレイヤーの部分を実装していきます。
サンプルコード全体

1. データモデル

要件の内、データモデルに該当する部分を型として実装していきます。言い換えると、一連のワークフローの中で、処理対象のデータが遷移し得る状態を定義していきます。
pythonですと、dataclassを使用して実現するのが便利と感じています。

# happy path
@dataclass(frozen=True)
class UnverifiedOrder: # 注文書
    item_id: str
    quantity: int
    delivery_method: DeliveryMethod
    shipping_to: CustomerAddress | ConvenienceStore

@dataclass(frozen=True)
class VerifiedOrder: # 確認済み注文書
    item_id: str
    quantity: Quantity
    shipping_to: CustomerAddress | ConvenienceStore

@dataclass(frozen=True)
class Invoice: # 請求書
    item_id: str
    quantity: Quantity
    total_price: Decimal
    shipping_to: CustomerAddress | ConvenienceStore

@dataclass(frozen=True)
class ShippedInvoice: # 発送票付き請求書
    total_price: Decimal
    shipping_to: CustomerAddress | ConvenienceStore
    arrival_date: datetime

# error path
@dataclass(frozen=True)
class OrderError: # エラー報告書
    code: str
    message: str

まんま以下の部分です。

2. 書類とその項目

2.2 確認済み注文書

• 記載項目: 商品ID、数量、配送先住所。

2.3 請求書

• 記載項目: 商品ID、数量、合計価格、配送先住所。

2.4 発送票付き請求書

• 記載項目: 合計価格、配送先住所、到着予定日。

2.5 エラー報告書

• 記載項目: エラーコード、エラーメッセージ。

すいません、正直コードを書いてから、自然言語の文章の方を寄せたからです。
現実の業務であれば、自然言語の文章に散らばっている要件を集めてエンコーディングしていくことになると思います。

しかし、前述の「商品受注業務マニュアル」もコードからリバースしたにしては、わりにマニュアル然とした、ありがちな構成になっているのではないでしょうか？

1. 全体の流れ/概要/総論を書き、
2. フローの中で登場してくる各データ(この場合書類)を示し、
3. 具体的なステップ/各論 を書いていく

という構成は、ベーシックであり理解し易い構造と思います。
その構造をそのままコードに持ってくれば、情報として分かりやすかろう(読みやすかろう)という考えがあります。

2. ビジネスロジック

ワークフローそのもの、あるいはそれぞれのタスクをエンコードします。pythonコードにおいては、関数に対応します。
まず、ワークフロー全体

# worlflow
def process_order(
    address_checker: Callable[[str, str], bool],
    product_catalog: Callable[[str], Decimal],
    lookup_delivery_days: LookUpDeliveryDaysMethods,
) -> Callable[[OrderProtocol], Result[ShippedInvoice, OrderError]]:

    def _process_order_core(
        order: OrderProtocol
    ) -> Result[ShippedInvoice, OrderError]:

        return (
            From(cast(UnverifiedOrder, order))
            .bind(review_order(address_checker))
            .bind(calculate_price(product_catalog))
            .bind(determine_arrival_date(lookup_delivery_days))
        )

    return _process_order_core

上下の部分(注入される依存性、IされるDを待ち受けるための部分です。)はひとまずスルーして、_process_order_coreに着目してください。

    def _process_order_core(
        order: OrderProtocol　 # VerifiedOrderと同等の構造。コールサイトを具体的なVerifiedOrderに依存させないための緩衝材（インターフェース）
    ) -> Result[ShippedInvoice, OrderError]:

1.1 「商品受注」全体のinputと成果物

• input: フロント業務より受け取った書類「注文書」
• 成果物: 発送票付き請求書 または 不備報告書

というドメイン上の事実(仕事には、インプットと成果物が存在します。)を、ワークフロー全体関数のシグネチャとしてエンコードしています。
Result[]の説明や、その実装については後述します。（中身は成功か失敗のいずれかである、という決まりを持った封筒に相当するものとイメージいただければと思います。）

From(cast(UnverifiedOrder, order))
.bind(review_order(address_checker))
.bind(calculate_price(product_catalog))
.bind(determine_arrival_date(lookup_delivery_days))

こちらは、以下の要件を表現します。
コード上も各ステップの記述に先んじて登場し、ヘッダー的・概要説明的な役割を果たします。

1.2 全体の動きの流れ

初期注文受付
-> 注文内容のチェック
-> 価格の計算
-> 配送日の決定

.bind()については後述します。

ワークフロー中の各タスクについては長いので、「価格の計算」タスクのみひとまず取り上げます。

def calculate_price(
    product_catalog: Callable[[str], Decimal]
) -> Callable[[VerifiedOrder], Result[Invoice, OrderError]]:

    def _calculate_price_core(order: VerifiedOrder) -> Result[Invoice, OrderError]:

        try:
            item_price = product_catalog(order.item_id)
        except KeyError:
            return Err(
                OrderError(
                    code="ItemNotFound",
                    message=f"The item_id {order.item_id} is not found in the product catalog.")
            )
        return Ok(
            Invoice(
                item_id=order.item_id,
                quantity=order.quantity,
                shipping_to=order.shipping_to,
                total_price=int(order.quantity) * item_price,
            )
        )

    return _calculate_price_core

まず、関数のシグネチャで

3.3 価格の計算

• input: 検証済み注文書
• 成果物: 請求書

を表し、関数の中身はもちろん実際のプロセスをコーディングします。

• 仕事: 封筒から書類を取り出し、商品情報に基づき価格を決定し、請求書を作成。商品情報が不足している場合は不備報告書を作成。書類を封筒に入れ、後続の係に渡す。

しっかり型をつけた関数のシグネチャ - 処理の中身 という構成も、
ヘッダー - 詳細(抽象 - 具体) という読む側に優しい構造の一種と考えます。

また、product_catalogの具体的な中身はまだ考えません。文字列を渡したら金額を返して欲しいというインターフェースの契約だけしておきます。
product_catalog: Callable[[str], Decimal]

Result[]、 Ok、Errという登場人物は、次の項で説明します。

共通化について

先送りにしてきた、bind()、Result[]、 Ok、Errの説明でもあります。

Resultは所謂Resultモナド（を模したもの）になります。関数型言語でなくとも、Rust(型としてはkotlin、swiftも)にビルドインされてますね。またGo言語で正常と異常の結果をタプルで返すアレともノリが近いですよね。

具体的な説明の前に度々すみませんが、もう一度「業務アニュアル」を見てください。

商品受注業務マニュアル>各タスクの手順

3. 各タスクの手順

3.1 初期注文受付

• フロントから注文書を受け取り、この$\textcolor{green}{書類を封筒に入れ、後続の係に渡す。}$

3.2 注文内容のチェック

• input: 注文書
• 成果物: 検証済み注文書
• 仕事: $\textcolor{green}{封筒から書類を取り出し}$、住所の有効性と数量の妥当性をチェック。不備がある$\textcolor{green}{場合は不備報告書を作成。書類を封筒に入れ、後続の係に渡す。}$

3.3 価格の計算

• input: 検証済み注文書
• 成果物: 請求書
• 仕事: $\textcolor{green}{封筒から書類を取り出し}$、商品情報に基づき価格を決定し、請求書を作成。商品情報が不足している$\textcolor{green}{場合は不備報告書を作成。書類を封筒に入れ、後続の係に渡す。}$

3.4 配送日の決定

• input: 請求書
• 成果物: 発送票付き請求書
• 仕事: $\textcolor{green}{封筒から書類を取り出し}$、配送先に基づき配送日数を計算し、発送票付き請求書 に記載。配送不可能な地域の$\textcolor{green}{場合は不備報告書を作成。書類を封筒に入れ、後続の係に渡す}$。

• わざわざ緑で強調している部分ですが、明らかに共通化/括り出し出来る働きがあります。
  • 「封筒(なんらかの文脈、構造)から書類(値)を取り出し、処理し、また封筒(なんらかの文脈、構造)に入れる働き」が明らかに共通化できる。また、「前後にその動き(封筒出し入れ)を差し込みつつ、タスクを逐次的に処理していく働き。」
    -> このような構造/操作DRYしたというのがモナド(の一面、ひとつの説明の仕方ではないか)
  • 「タスクの中で不備があれば不備報告書を、そうでなければ通常の書類を封筒に入れる働き」、言い換えると「封筒の中身は正常時の書類or失敗時の書類が入っているという文脈」を共通化できる。
    -> このような文脈をDRYしたのがResult
• この２点の横断的関心事を共通化したものがResultのモナド、Resultモナド（の一面）だと理解しています。
  より具体的に言えば、if hoge != null else　なんちゃら処理``if hoge != null elseなんちゃら処理if hoge != null else　なんちゃら処理のようなテンプレ構造を共通化し、メインのフローの記述から分離したいものとも言えましょうか。

用語のアカデミックな、厳密な定義を気にされる方はこちらをご確認ください。

上記の横断的な要件を括り出したユーティリティ的なコードが以下です。

result.py

from dataclasses import dataclass
from typing import Any, Callable, TypeVar

T = TypeVar('T')
E = TypeVar('E')
U = TypeVar('U')

@dataclass(frozen=True)
class Ok[T]:
    value: T

    def bind(self, op: Callable[[T], 'Result[U, E]']) -> 'Result[U, E]':
        return op(self.value)

    def or_else(self, op: Callable[[Any], 'Result[T, E]']) -> 'Result[T, E]':
        return self

@dataclass(frozen=True)
class Err[E]:
    error: E

    def bind(self, op: Callable[[Any], 'Result[U, E]']) -> 'Result[U, E]':
        return self

    def or_else[F](self, op: Callable[[Any], 'Err[F]']) -> 'Err[F]':
        return op(self.error)

type Result[T, E] = Ok[T] | Err[E]

def From(value: T) -> Result[T, Any]:
    return Ok(value)

いくつか要素がありますが、Resultが本体で、しかしその実体は、OkまたはErrいずれかである。そして、その両者がbind()を持つという構成です。

type Result[T, E] = Ok[T] | Err[E]

封筒の中身は正常時の書類or失敗時の書類が入っているという文脈

bind()メソッドは、Result型を戻り値とする関数を受けとり、その関数に自身の値を適用します。

封筒から書類を取り出し、(コード上は引数として受け取った)処理をし、また封筒に入れ次の処理に繋ぐ働き

それにより、以下のように純粋関数をチェーンしてResultに内包されたデータをコロコロ転がしていく記載ができます。
UNIXのパイプや、リスト構造によく生えてる.map(より近いのはflatMap)のようなものですかね。

	From(cast(UnverifiedOrder, order))
	.bind(review_order(address_checker))
	.bind(calculate_price(product_catalog))
	.bind(determine_arrival_date(lookup_delivery_days))

また、Resultの中身がErrだった場合(Errに生えてる方のbind())は、何もせずスルーします。

タスクの各係は、封筒の中身が不備通知書であったら、なにもせずそっと封筒に戻して後続の係に渡す

    def bind(self, op: Callable[[Any], 'Result[U, E]']) -> 'Result[U, E]':
        return self

参考Railway Oriented Programming

これにより、

封筒(なんらかの文脈、構造)から書類(値)を取り出し、処理し、また封筒(なんらかの文脈、構造)に入れる働き。前後にその動きを差し込みつつ、タスクを逐次的に処理していく働き。

を共通化できました。
また、実装面の話をすると、エラー処理は基本的にResultを使用していきますが、例外とは以下の方針で使い分けることとします。
- ドメインエラー: ビジネス上あり得る状態であり、想定内のエラー。業務上のエラー。ドメインとモデルの一種としてのエラー
- 例外: 想定外のエラー。言葉通りの例外。主にシステムインフラのエラー。panic。
- ex) ゼロ除算をしてクラッシュした。サーバが落ちた。

また、エラーを例外ではなくResultで表現すると、関数のシグネチャで失敗するということ、どのような失敗を返すかがおおまかにわかり親切ですね。

3. ビジネスルール

代数的データ型

さて、開発を進めている内に、顧客は新たなビジネスアイディアを取り入れることを決定しました。

エンドユーザは、商品の配送方法として 1.自宅受け取り と 2.コンビニ受け取り を選択できる。
自宅受け取りの場合は、情報として 顧客住所 が必要であり、コンビニ受け取りの場合は コンビニチェーン名と店舗コードが必要である。

この要望を受け入れ、実はすでに実装していました。

@dataclass(frozen=True)
class UnverifiedOrder: # 確認済み注文書
    item_id: str
    quantity: int
    shipping_to: CustomerAddress | ConvenienceStore # <-特にここの部分

ユニオンタイプにより、「そのまま」要件をエンコーディングしていると感じます。

type Franchisors = Literal["SevenEleven", "FamilyMart", "Lawson"]

@dataclass(frozen=True)
class CustomerAddress:
    prefecture: str
    detail: str

@dataclass(frozen=True)
class ConvenienceStore:
    company: Franchisors
    store_code: str

送付先が自宅かコンビニかによって、各タスクの内容が異なります。

def review_order(
        check_address_existence: Callable[[str, str], bool]
    ) -> Callable[[UnverifiedOrder], Result[VerifiedOrder, OrderError]]:

    def _with_specific_check_method(
            order: UnverifiedOrder
    ) -> Result[VerifiedOrder, OrderError]:

        match order.shipping_to: # 送付先が自宅かコンビニかによってサブタスクが異なる
            case CustomerAddress(prefecture=pref, detail=det):
                if not check_address_existence(pref, det): # 日本の住所の実在チェック
                    return Err(
                        OrderError(code="InvalidAddress",message="The provided address is invalid.")
                    )
            case ConvenienceStore(company=_, store_code=code): # コンビニ店舗のの実在チェック
                if code == "": # 実際の処理はダミー
                    return Err(
                        OrderError(code="InvalidStoreCode",message="The provided store code is invalid.")
                    )

値オブジェクトとスマートコンストラクタ

更に、以下のような個別のビジネスルールも存在しています。

商品の注文「数量」は、0であることはあり得ず、99個までしか受け付けない。

スマートコンストラクタ(Fromメソッド)を実装し、pythonに翻訳しました。

# Value Object Example
@dataclass(frozen=True)
class Quantity:
    value: int

    def __post_init__(self):
        if not 1 <= self.value <= 99:
            raise ValueError("Quantity must be between 1 and 99")

    @staticmethod
    def From(value: int) -> Result['Quantity', str]:
        try:
            return Ok(Quantity(value))
        except ValueError as e:
            return Err(str(e))

    def __int__(self):
        return self.value

Fromメソッドは、結果をResultで返却します。

        return Quantity.From(order.quantity).bind(
            lambda quantity: Ok( # ビジネスルール適合時
                VerifiedOrder(
                    item_id=order.item_id,
                    quantity=quantity,
                    shipping_to=order.shipping_to
                )
            )
        ).or_else( # ビジネスルール違反時のフォールバック処理。デフォルト値を書くようなことも可能
            lambda error: Err(
                OrderError(code="Order", message=error)
            )
        ) 

Javaだったりのように、デフォルトコンストラクタを使用できないようにする方法は発見できませんでした。From()を使用せずに生成すると、例外を投げる可能性があります。

4. データアクセス

さて、ここで「商品受注業務」を更に観察していると、以下の事実が明らかになりました。

商品受注チームには、チームリーダと、各タスクを行うメンバー達が存在し、それぞれ役割を明確に分担している。
チームリーダ: フロント業務とのやりとり、後段の業務チーム(商品発送チーム)への通知、インプット以外の業務に必要なデータ(依存している情報)を取得、チームに渡す。
ex)例えば、商品カタログを倉庫から取ってきて、メンバーに業務指示と共に渡すイメージ
メンバ: リーダからインプットと業務遂行に必要なデータ、業務指示を受け取り、実際のタスクを実施する。

これは、以下のような狙いがあるようです。

• リーダに外部とのやりとりする機能や外部からデータを取得する権限を集約する
• 逆にメンバーは、本質的なタスク実行そのものに集中出来る。(外部とのやりとりや、データ取得方法の詳細を気にしなくていい)

無理に喩えずに、実装の話をそのまました方がいいかもしれませんね。
要はテスト容易性などのために、コードの純粋な部分を非純粋な部分から分離したいということです。
いつものオニオンの外側を今回はservice層（画像で言ったらinfrastructure+application serviceのイメージ）、内側をworkflow層とします。

• service層
  • 非純粋(副作用コードあり、例外も投げる、FW依存機能を使う)
  • httpのやりとり、データアクセスロジックの取得/注入、他のサービスへのイベントpush
  • 今まで見てきたコードをコールする側のコードでありエンドポイントを提供する部分のコード
• worlflow層
  • 純粋関数で構成されるビジネスロジック
  • 今まで見てきたコードはこの層のコード

workflow層も、完全に純粋な関数にはなっていないです。

今回の考え方

• 実際に楽になるvs新しい概念を導入するハードル のコスパが良いことをする
  • IOモナド等は導入しない。「テストが書きやすくなる」程度の純粋さを目指す
  • workflow層内部で、渡されたreadの外部IOを行うことは許容する。（テスト時は、IOアクションのモック関数を渡すことでまあまあ書きやすいと思うため）
  • write系の実行は、モック的な工夫が複雑になるためworkflow層外に押しやる

本題の、データアクセスについては以下のようにコーディングしました。
workflow層

# data access requirement inerfaces
type ToHome = Callable[[str], int | None]
type ToCVS  = Callable[[str, str], int]
type LookUpDeliveryDaysMethods = tuple[ToHome, ToCVS]

# workflow
def process_order( # ↓ データアクセスロジックの注入。（DをIする）
    address_checker: Callable[[str, str], bool],
    product_catalog: Callable[[str], Decimal],
    lookup_delivery_days: LookUpDeliveryDaysMethods,
) -> Callable[[OrderProtocol], Result[ShippedInvoice, OrderError]]:

    def _process_order_core(
        order: OrderProtocol
    ) -> Result[ShippedInvoice, OrderError]:

        return (
            From(cast(UnverifiedOrder, order))
            .bind(review_order(address_checker)) # 各タスクに適用する。タスク側では、渡されたロジックを実行する。
            .bind(calculate_price(product_catalog))
            .bind(determine_arrival_date(lookup_delivery_days))
        )

    return _process_order_core

service層

# FastAPIでエンドポイントを実装している
@router.post("", operation_id="create_order", response_model=OrderResponse)
async def create_order(
    order: Annotated[ # swaggerの定義のための部分
        OrderRequest,
        Body(
            openapi_examples={
                "home_delivery": home_delivery_example,
                "convenience_store_delivery": convenience_store_delivery_example,
            }
        ),
    ]
) -> OrderResponse:

    order_workflow = process_order( # wokflow関数に適用
        existence_check_japanese_address, # 具体的なデータアクセスロジック
        product_catalog,
        lookup_delivery_days_pack,
    )

    match order_workflow(order):
        case Ok(o):
            ordered_event = OrderResponse(
                    bill_amount=o.total_price,
                    arrival_date=o.arrival_date,
                    shipping_to=o.shipping_to,
                )
            send_event(order_response_to_json(ordered_event)) # 他のサービスへのイベント通知。orderedイベントログをkey-valueストアに永続化したい場合のロジックがあればこの辺に各想定。
            return ordered_event # httpレスポンス
        case Err(e): # エラー時のhttpレスポンス
            raise HTTPException(status_code=400, detail=e.message)
        case _:
            raise HTTPException(status_code=500, detail="unexpected error in http layer")

実際のデータアクセスロジックは適当なモックです。

以上でひととおりのトピックを説明しました。
説明が分かりにくい箇所もあると思いますので、繰り返しですが本格的な理解には
Domain Modeling Made Functional
をお勧めします。(和訳出て欲しい。)

終わり、雑感

本格的に関数型プログラミングに習熟したわけではありませんが、Domain Modeling Made Functionalにも言及があるとおり、業務ロジックやwebサービスバックエンドロジックの要件をエンコーディングすることに向いたアプローチではないかと思いました。
ゲーム等だとまた前提が異なるかもしれませんが、本質的にはIn/Outがあるタスクの連続した逐次処理であるようなドメインのモデリングはかなり直接的に要件を表現出来るのではないかなと思います。

また、pythonで実現するにあたってResultの不在(ビルドインやメジャーライブラリ)がネックかと考えていましたが、自前実装してもシンプルなものに止まるということが意外でした。(使い込んでないので不備があればご指摘ください。)
この程度のシンプルさであれば、「みんなよくわからない複雑な機構を持ち込むな」という話にもならないのではないかな、と思います。

今回は、

• Resultモナド: エラー制御フローのドメインロジックからの分離、純粋関数のパイプライン構築
• ADT: 要件の直感的なモデリング、セルフドキュメンテーション、型によるコンパイルタイムのフィードバック、IDEの補完

など、概念の導入ハードル/コスト vs 実際のリターン

のコスパが良い（主観で）と思う要素のみ取り入れました。本格的にやるならもちろんscalaやF#等でやらないと厳しいとは思います。(kotlinやTypescriptに関数型ライブラリを補完するアプローチもありがちでしょうか？)

脚注

1. ドメインロジックのいわば幹の部分であり、本筋の仕事の流れを表す部分のニュアンスで使用しています。

   • ex) 各タスクをエンコードした関数そのもの

2. ドメインロジックのいわば枝葉の部分であり、判断条件やバリデーションなど、個別のルールの部分というニュアンスで使用しています。

   • 値オブジェクトの持つバリデーション、型による値の制約、分岐を判断するロジックなど