error_handling_pipelines.py

from dataclasses import dataclass, replace
from joblib import Parallel, delayed
from typing import Optional, Callable, List, Iterable

@dataclass(frozen=True)
class Either:
    value: Optional = None
    exception: Optional = None
        
def forward_exceptions(fun: Callable):
    def wrapped(arg):
        if not isinstance(arg, Either):
            arg = Either(value=arg)
        if arg.exception:
            return
        try:
            return replace(arg, value=fun(arg.value))
        except Exception as ex:
            return replace(arg, value=None, exception=ex)
    return wrapped

@forward_exceptions
def length(x: str) -> int:
    return len(x)

@forward_exceptions
def add1(x: int) -> int:
    return x + 1

@forward_exceptions
def outcome(x: int) -> str:
    return "win" if x > 5 else "loose"

def compose(*functions):
    def composed(x):
        for f in functions:
            x = f(x)
        return x
    return composed

def pmap(fun: Callable, xs: Iterable):
    return Parallel(32)(delayed(fun)(x) for x in xs)

xs = ("cart", "intention", "wander", 1, "iron", "chalk")
pipeline = compose(length, add1, outcome)
ys = pmap(pipeline, xs)

"""
>>> ys
[Either(value='loose', exception=None),
 Either(value='win', exception=None),
 Either(value='win', exception=None),
 Either(value=None, exception=TypeError("'>' not supported between instances of 'NoneType' and 'int'")),
 Either(value='loose', exception=None),
 Either(value='win', exception=None)]
"""