Async-Option monad composition revised

async_option.py

import asyncio

from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from typing import Any


def curry[First, *Rest, Result](function: Callable[[First, *Rest], Result]) -> Callable[[*Rest], Callable[[First], Result]]:
    return lambda *rest: lambda first: function(first, *rest)


@curry
@curry
@curry
def bind[From, MFrom, MTo, MMTo](m_value: MFrom, function: Callable[[From], MTo], join: Callable[[MMTo], MTo], map: Callable[[Callable[[From], MTo]], Callable[[MFrom], MMTo]]) -> MTo:
    return join(map(function)(m_value))


@curry
@curry
@curry
def compose[From, Intermediate, MIntermediate, MTo](value: From, first: Callable[[From], MIntermediate], second: Callable[[Intermediate], MTo], bind: Callable[[Callable[[Intermediate], MTo]], Callable[[MIntermediate], MTo]]) -> MTo:
    return bind(second)(first(value))


def identity_identity[T](value: T) -> T:
    return value


@curry
def identity_map[From, To](value: From, function: Callable[[From], To]) -> To:
    return function(value)


identity_join = identity_identity


identity_bind = identity_map


identity_compose = compose(identity_bind)


type Option[T] = None | Some[T]


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


def option_identity[T](value: T) -> Option[T]:
    return Some(value)


@curry
def option_map[From, To](option_value: Option[From], function: Callable[[From], To]) -> Option[To]:
    match option_value:
        case None:
            return None
        case Some(value):
            return Some(function(value))


@curry
def option_inject[T, MOptionT](option_m_option_value: Option[MOptionT], m_identity: Callable[[Option[T]], MOptionT]) -> MOptionT:
    match option_m_option_value:
        case None:
            return m_identity(None)
        case Some(m_option_value):
            return m_option_value


option_join = option_inject(identity_identity)


option_bind = bind(option_map)(option_join)


option_compose = compose(option_bind)


type Async[T] = Coroutine[Any, Any, T]


async def async_identity[T](value: T) -> T:
    return value


@curry
async def async_map[From, To](async_value: Async[From], function: Callable[[From], To]) -> To:
    return function(await async_value)


async def async_join[T](async_async_value: Async[Async[T]]) -> T:
    return await (await async_async_value)


async_bind = bind(async_map)(async_join)


async_compose = compose(async_bind)


async_option_identity = identity_compose(async_identity)(option_identity)


async_option_map = identity_compose(async_map)(option_map)


async_option_inject = option_inject(async_identity)


async_option_join = async_bind(async_option_inject)


async_option_bind = bind(async_option_map)(async_option_join)


async_option_compose = compose(async_option_bind)


async def root(number: float) -> Option[float]:
    return Some(number ** .5) if number >= 0 else None


async def reciprocal(number: float) -> Option[float]:
    return Some(1 / number) if number else None


root_of_reciprocal = async_option_compose(root)(reciprocal)


def main() -> None:
    number = float(input('Enter a number: '))
    result = asyncio.run(root_of_reciprocal(number))
    print(result)


main()