image-classification-service/image_prediction/utils/generic.py

from functools import wraps
from inspect import signature
from itertools import starmap
from typing import Callable

from funcy import iterate, first, curry, map
from pymonad.either import Left, Right, Either
from pymonad.tools import curry as pmcurry

from image_prediction.utils import get_logger

logger = get_logger()


def until(cond, func, *args, **kwargs):
    return first(filter(cond, iterate(func, *args, **kwargs)))


def lift(fn):
    return curry(map)(fn)


def starlift(fn):
    return curry(starmap)(fn)


def bottom(*args, **kwargs):
    return False


def top(*args, **kwargs):
    return True


def left(fn):
    @wraps(fn)
    def inner(x):
        return Left(fn(x))

    return inner


def right(fn):
    @wraps(fn)
    def inner(x):
        return Right(fn(x))

    return inner


def wrap_left(fn, success_condition=top, error_message=None) -> Callable:
    return wrap_either(Left, Right, success_condition=success_condition, error_message=error_message)(fn)


def wrap_right(fn, success_condition=top, error_message=None) -> Callable:
    return wrap_either(Right, Left, success_condition=success_condition, error_message=error_message)(fn)


def wrap_either(success_type, failure_type, success_condition=top, error_message=None) -> Callable:
    @wraps(wrap_either)
    def wrapper(fn) -> Callable:

        n_params = len(signature(fn).parameters)

        @pmcurry(n_params)
        @wraps(fn)
        def wrapper(*args, **kwargs) -> Either:
            try:
                result = fn(*args, **kwargs)
                if success_condition(result):
                    return success_type(result)
                else:
                    return failure_type({"error": error_message, "result": result})
            except Exception as err:
                logger.error(err)
                return failure_type({"error": error_message or err, "result": Void})

        return wrapper

    return wrapper


class Void:
    pass