pythonでResultモナド

python 3.12

result.py

from typing import Callable, TypeVar, Any

from dataclasses import dataclass


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

U = TypeVar('U')

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


class Result[T, E]:
    def __init__(self, value: Ok[T] | Err[E]) -> None:
        self._container = value


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

    @classmethod
    def Ok(cls: type['Result[T, Any]'], value: T):
        return cls(Ok(value))

    @classmethod
    def Err(cls: type['Result[Any, E]'], value: E):
        return cls(Err(value))

    def unwrap(self):
        return self._container

yoda keisuke

使用側
res.unwrap()
とするのがやだ。
class Result[T, E]:
def init(self, value: Ok[T] | Err[E]) -> None:
self._container = value
としなければならず、ラッピングレイヤーが無駄に１つ被さっている感じがする。

from app.common.result import Result, Ok, Err

# 各ステップの中身は割愛
res =  Result.Ok(cast(UnverifiedOrder, order))
            .bind(review_order(address_checker))
            .bind(calculate_price(product_catalog))
            .bind(determine_arrival_date_for_japan)

match res.unwrap():
        case Ok(value):  # resultにis_okとか生やした方が良い？
            return(
                OrderResponse(
                        item_id=value.item_id,
                        bill_amount=value.total_price,
                        arrival_date=value.arrival_date,
                )
            )
        case Err(error):
            raise HTTPException(status_code=400, detail=error.message)

yoda keisuke

これがいいのではないか。
使用側のimportで

from app.common.util.result import Err, From, Ok, Result

するのがちょっとやだ。Result.Fromみたいにしたい

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]':
        print("Ok bind")
        return op(self.value)

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

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

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

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

yoda keisuke

IOモナドっぽくできる..?

import asyncio

class MyFuture(asyncio.Future):
    async def flatMap(self, func):
        result = await self
        return await func(result)

async def async_function1(x):
    await asyncio.sleep(1)
    return x * 2

async def async_function2(y):
    await asyncio.sleep(1)
    return y + 3

async def main():
    future1 = MyFuture()
    asyncio.create_task(async_function1(5), loop=future1._loop)

    future2 = await future1.flatMap(async_function2)
    result = await future2
    return result

# 非同期処理の実行
result = asyncio.run(main())
print(f"The result is: {result}")