Awaitable-Option monad composition

awaitable_option.py

from asyncio import run
from collections.abc import Coroutine
from dataclasses import dataclass
from typing import Any, Protocol


class Callable[*Arguments, Result](Protocol):
    def __call__(self, *args: *Arguments) -> Result:
        ...


@dataclass(frozen=True)
class Partial1[First, *Rest, Result]:
    function: Callable[First, *Rest, Result]
    first: First

    def __call__(self, *args: *Rest) -> Result:
        return self.function(self.first, *args)


@dataclass(frozen=True)
class Partial2[First, Second, *Rest, Result]:
    function: Callable[First, Second, *Rest, Result]
    first: First
    second: Second

    def __call__(self, *args: *Rest) -> Result:
        return self.function(self.first, self.second, *args)


def identity[Value](value: Value) -> Value:
    return value


def compose[From, Intermediate, To](second: Callable[Intermediate, To], first: Callable[From, Intermediate], value: From) -> To:
    return second(first(value))


def make_compose[From, Intermediate, To](second: Callable[Intermediate, To], first: Callable[From, Intermediate]) -> Callable[From, To]:
    return Partial2(compose, second, first)


@dataclass(frozen=True)
class Nothing:
    pass


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


type Option[Value] = Nothing | Some[Value]


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


def option_map[From, To](function: Callable[From, To], option: Option[From]) -> Option[To]:
    match option:
        case Nothing():
            return Nothing()
        case Some(value):
            return Some(function(value))


def make_option_map[From, To](function: Callable[From, To]) -> Callable[Option[From], Option[To]]:
    return Partial1(option_map, function)


def option_join[Value](option_option: Option[Option[Value]]) -> Option[Value]:
    match option_option:
        case Nothing():
            return Nothing()
        case Some(option):
            return option


def make_option_bind[From, To](function: Callable[From, Option[To]]) -> Callable[Option[From], Option[To]]:
    return make_compose(option_join, make_option_map(function))


def option_bind[From, To](function: Callable[From, Option[To]], option: Option[From]) -> Option[To]:
    return make_option_bind(function)(option)


def make_option_compose[From, Intermediate, To](second: Callable[Intermediate, Option[To]], first: Callable[From, Option[Intermediate]]) -> Callable[From, Option[To]]:
    return make_compose(make_option_bind(second), first)


def option_compose[From, Intermediate, To](second: Callable[Intermediate, Option[To]], first: Callable[From, Option[Intermediate]], value: From) -> Option[To]:
    return make_option_compose(second, first)(value)


type Awaitable[Value] = Coroutine[Any, Any, Value]


async def awaitable_identity[Value](value: Value) -> Value:
    return value


async def awaitable_map[From, To](function: Callable[From, To], awaitable: Awaitable[From]) -> To:
    return function(await awaitable)


def make_awaitable_map[From, To](function: Callable[From, To]) -> Callable[Awaitable[From], Awaitable[To]]:
    return Partial1(awaitable_map, function)


async def awaitable_join[Value](awaitable_awaitable: Awaitable[Awaitable[Value]]) -> Value:
    return await (await awaitable_awaitable)


def make_awaitable_bind[From, To](function: Callable[From, Awaitable[To]]) -> Callable[Awaitable[From], Awaitable[To]]:
    return make_compose(awaitable_join, make_awaitable_map(function))


def awaitable_bind[From, To](function: Callable[From, Awaitable[To]], awaitable: Awaitable[From]) -> Awaitable[To]:
    return make_awaitable_bind(function)(awaitable)


def make_awaitable_compose[From, Intermediate, To](second: Callable[Intermediate, Awaitable[To]], first: Callable[From, Awaitable[Intermediate]]) -> Callable[From, Awaitable[To]]:
    return make_compose(make_awaitable_bind(second), first)


def awaitable_compose[From, Intermediate, To](second: Callable[Intermediate, Awaitable[To]], first: Callable[From, Awaitable[Intermediate]], value: From) -> Awaitable[To]:
    return make_awaitable_compose(second, first)(value)


def awaitable_option_identity[Value](value: Value) -> Awaitable[Option[Value]]:
    return awaitable_identity(option_identity(value))


def awaitable_option_inject[Value](option_awaitable_option: Option[Awaitable[Option[Value]]]) -> Awaitable[Option[Value]]:
    match option_awaitable_option:
        case Nothing():
            return awaitable_identity(Nothing())
        case Some(awaitable_option):
            return awaitable_option


def make_awaitable_option_map[From, To](function: Callable[From, To]) -> Callable[Awaitable[Option[From]], Awaitable[Option[To]]]:
    return make_awaitable_map(make_option_map(function))


def awaitable_option_map[From, To](function: Callable[From, To], awaitable_option: Awaitable[Option[From]]) -> Awaitable[Option[To]]:
   return make_awaitable_option_map(function)(awaitable_option)


def awaitable_option_join[Value](awaitable_option_awaitable_option: Awaitable[Option[Awaitable[Option[Value]]]]) -> Awaitable[Option[Value]]:
    return awaitable_bind(awaitable_option_inject, awaitable_option_awaitable_option)


def make_awaitable_option_bind[From, To](function: Callable[From, Awaitable[Option[To]]]) -> Callable[Awaitable[Option[From]], Awaitable[Option[To]]]:
    return make_compose(awaitable_option_join, make_awaitable_option_map(function))


def awaitable_option_bind[From, To](function: Callable[From, Awaitable[Option[To]]], awaitable_option: Awaitable[Option[From]]) -> Awaitable[Option[To]]:
    return make_awaitable_option_bind(function)(awaitable_option)


def make_awaitable_option_compose[From, Intermediate, To](second: Callable[Intermediate, Awaitable[Option[To]]], first: Callable[From, Awaitable[Option[Intermediate]]]) -> Callable[From, Awaitable[Option[To]]]:
    return make_compose(make_awaitable_option_bind(second), first)


def awaitable_option_compose[From, Intermediate, To](second: Callable[Intermediate, Awaitable[Option[To]]], first: Callable[From, Awaitable[Option[Intermediate]]], value: From) -> Awaitable[Option[To]]:
    return make_awaitable_option_compose(second, first)(value)


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


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


root_of_reciprocal = make_awaitable_option_compose(root, reciprocal)


print(run(root_of_reciprocal(float(input()))))