ROP

railway_oriented_programming.py

import asyncio
from functools import reduce
from typing import Any, Callable, Generic, TypeVar, Union

T = TypeVar("T")
R = TypeVar("R")


class Ok(Generic[T]):
    def __init__(self, value: T) -> None:
        self.value = value


class Error(Generic[R]):
    def __init__(self, value: R) -> None:
        self.value = value


TwoTrack = Union[Ok[T], Error[R]]

OneTrackFunction = Callable[[T], TwoTrack[T, R]]
TwoTrackFunction = Callable[[TwoTrack[T, R]], TwoTrack[T, R]]


def flat_map(fn):
    def wrapper(v):
        if isinstance(v, Error):
            return v
        return fn(v)

    async def async_wrapper(v):
        if isinstance(v, Error):
            return v
        return await fn(v)

    return async_wrapper if asyncio.iscoroutinefunction(fn) else wrapper


def pipe(*funcs: Callable[[T], R]) -> Callable[[T], R]:
    return lambda value: reduce(lambda v, f: f(v), funcs, value)

def async_pipe(*funcs):
    async def fn(init_v):
        loop = asyncio.get_running_loop()
        return reduce(lambda f, v: loop.run_until_complete(f(v)), funcs, init_v)    
    return fn


def async_pipe(*funcs):
    async def fn(init_v):
        def await_if_coro(f, v):
            loop = asyncio.get_event_loop()
            if asyncio.iscoroutinefunction(f):
                return loop.run_until_complete(f(v))
            return f(v)
        return reduce(lambda v, f: await_if_coro(f, v), funcs, init_v)
    return fn