Python helpers to make working with `Result`s more like python

python_cope.py

from typing import Concatenate, NoReturn, overload, NamedTuple, Any
from collections.abc import Callable
from returns.result import Result, Success, Failure
from returns.pipeline import is_successful


type EarlyReturnF[R] = Callable[[R], NoReturn]


def early_return_exc[**Spec, R](
    func: Callable[Concatenate[EarlyReturnF[R], Spec], R],
) -> Callable[Spec, R]:

    def wrapped(*args, **kwargs):
        class EarlyReturn(BaseException):
            pass

        def early_return(return_value: R) -> NoReturn:
            raise EarlyReturn(return_value)

        try:
            return func(early_return, *args, **kwargs)
        except EarlyReturn as e:
            return e.args[0]

    return wrapped


class TryUnwrapOp[R, E](NamedTuple):
    return_f: EarlyReturnF[Result[R, E]]

    @overload
    def __call__[R_](self, res: Success[R_]) -> R_: ...

    @overload
    def __call__[R_](self, res: Failure[E]) -> NoReturn: ...

    @overload
    def __call__[R_](self, res: Result[R_, E]) -> R_ | NoReturn: ...

    def __call__[R_](self, res: Result[R_, E]) -> R_ | NoReturn:
        if not is_successful(res):
            self.return_f(res)
        return res.unwrap()


type AnyTryUnwrapOp[E] = TryUnwrapOp[Any, E]

type AnyTryUnwrapOpE = TryUnwrapOp[Any, Exception]

type TryUnwrapOpE[R] = TryUnwrapOp[R, Exception]


@early_return_exc
def test_early_return(
    return_: EarlyReturnF[Result[int, str]], a: int
) -> Result[int, str]:
    q = TryUnwrapOp(return_)
    value = q(Success(a) if not a > 10 else Failure("argument is too large"))
    return Success(value * 100)


@overload
def with_try[**Spec, S, F](
    func: Callable[Concatenate[AnyTryUnwrapOp[F], Spec], Result[S, F]],
) -> Callable[Spec, Result[S, F]]: ...


@overload
def with_try[**Spec, S, F](
    func: Callable[Concatenate[TryUnwrapOp[S, F], Spec], Result[S, F]],
) -> Callable[Spec, Result[S, F]]: ...


def with_try[**Spec, S, F](
    func: Callable[Concatenate[TryUnwrapOp[S, F], Spec], Result[S, F]],
) -> Callable[Spec, Result[S, F]]:
    def wrapped(*args, **kwargs):
        # making it a closure and valid per-call should make it less
        # likely to be misused.
        class EarlyReturn(BaseException):
            pass

        def early_return(return_value: Result[S, F]) -> NoReturn:
            raise EarlyReturn(return_value)

        try:
            return func(TryUnwrapOp(early_return), *args, **kwargs)
        except EarlyReturn as e:
            return e.args[0]

    return wrapped


def augment_exception(new_exc: Exception) -> Callable[[Exception], Exception]:
    """
    Returns an exception mapper that augments an exception by raising it with the
    `from` keyword. Instead of directly subsituting the old exception, this adds
    it to the trace so if raised it will contain all the useful information.
    """
    def augment(old_exc: Exception) -> Exception:
        try:
            raise new_exc from old_exc
        except Exception as e:
            return e

    return augment

# Here's an example use case where TryUnwrapOp is used as the ? operator in rust
""""
@with_try
def get_otras_liquidaciones(
    q: AnyTryUnwrapOpE,
    elements: ElementList | list[PDFElement],
) -> ResultE[tuple[list[OtrasLiquidaciones], BoundingBox] | None]:
    liq_e_matches = [e for e in elements if e.text().startswith("OTRAS LIQUIDACIONES")]
    if len(liq_e_matches) == 0:
        return Success(None)
    if len(liq_e_matches) > 1:
        return Failure(
            ParsingError(
                f"Found more than one text field matching text {'OTRAS LIQUIDACIONES'!r}"
            )
        )
    (otras_liq_e,) = liq_e_matches
    top_of_bb = otras_liq_e.bounding_box.y1
    elements_below = [e for e in elements if e.bounding_box.y1 < top_of_bb]
    (concepto_e, asig_e), rest = q(
        find_exact_matches(
            elements_below,
            [
                lambda e: e.text() == "CONCEPTO",
                lambda e: e.text() == "ASIG",
            ],
        ).alt(augment_exception(ParsingError("Failed to find all required headers")))
    )
    (descuento_e,), rest = q(
        try_find_exact_matches(
            rest,
            [lambda e: e.text() == "DESCUENTO"],
        ).alt(augment_exception(ParsingError("Failed to find optional headers")))
    )
    result = cluster_columns(
        (concepto_e, descuento_e, asig_e), rest, 6, lambda e: e.bounding_box
    )
""""