Package raffiot

@final
class ComputationStatus(IntEnum):
    FAILED = 0
    SUCCEEDED = 1

Ancestors

• enum.IntEnum
• builtins.int
• enum.Enum

Class variables

var FAILED

var SUCCEEDED

@final
@dataclass
class Errors(Result[E, A]):
    """
    The result of a computation that failed on an excepted normal errors case.
    The program is still in a valid state and can progress safely.
    """

    __slots__ = "errors"

    errors: List[E]

Ancestors

• Result
• typing.Generic

Instance variables

var errors : List[~E]

Return an attribute of instance, which is of type owner.

Inherited members

• Result:
  • flatten
  • is_error
  • is_ok
  • is_panic
  • raise_on_panic
  • to_computation_status
  • unsafe_ap
  • unsafe_catch
  • unsafe_flat_map
  • unsafe_fold
  • unsafe_fold_raise
  • unsafe_get
  • unsafe_map
  • unsafe_map_error
  • unsafe_map_panic
  • unsafe_recover
  • unsafe_tri_map
  • zip

@final
class IO(Generic[R, E, A]):
    """
    Represent a computation that computes a value of type A,
    may fail with an errors (expected failure) of type E and have access
    anytime to a read-only context of type R.

    /!\\ **VERY IMPORTANT** /!\\

    1. **DO NEVER SUB-CLASS IO**: it would break the API.
    2. **DO NEVER INSTANTIATE an IO DIRECTLY**: use **only** the functions
       ands methods in this module.
    3. The IO is **LAZY**:
        no code is run until you invoke the run method.
    4. The IO never raises exceptions (unless there is a bug):
        it returns panics instead.
    5. The IO is stack-safe, but you need to make sure your own code is too!
        use defer and defer_io to avoid stack-overflow.

    Have a look to the documentation and examples to learn how to use it.
    """

    __slots__ = "__tag", "__fields"

    def __init__(self, __tag, __fields):
        self.__tag = __tag
        self.__fields = __fields

    def map(self, f: Callable[[A], A2]) -> IO[R, E, A2]:
        """
        Transform the computed value with f if the computation is successful.
        Do nothing otherwise.
        """
        return IO(IOTag.MAP, (self, f))

    def flat_map(self, f: Callable[[A], IO[R, E, A2]]) -> IO[R, E, A2]:
        """
        Chain two computations.
        The result of the first one (self) can be used in the second (f).
        """
        return IO(IOTag.FLATMAP, (self, f))

    def then(self, *others: IO[R, E, A2]) -> IO[R, E, A2]:
        """
        Chain two computations.
        The result of the first one (self) is dropped.
        """
        if len(others) == 1 and isinstance(others[0], abc.Iterable):
            return IO(IOTag.SEQUENCE, list((self, *others[0])))
        return IO(IOTag.SEQUENCE, list((self, *others)))

    def zip(self: IO[R, E, A], *others: IO[R, E, X]) -> IO[R, E, Iterable[A]]:
        """
        Pack a list of IO (including self) into an IO computing the list
        of all values.

        If one IO fails, the whole computation fails.
        """
        if len(others) == 1 and isinstance(others[0], abc.Iterable):
            return IO(IOTag.ZIP, list((self, *others[0])))
        return IO(IOTag.ZIP, list((self, *others)))

    def zip_par(self: IO[R, E, A], *others: IO[R, E, A]) -> IO[R, E, List[A]]:
        """
        Pack a list of IO (including self) into an IO computing the list
        of all values in parallel.

        If one IO fails, the whole computation fails.
        """
        return zip_par(self, *others)

    def parallel(self: IO[R, E, A], *others: IO[R, E, A]) -> IO[R, E, List[Fiber]]:
        """
        Run all these IO (including self) in parallel.
        Return the list of fibers, in the same order.

        Each Fiber represent a parallel computation. Call

        >>> wait([fiber1, fiber2, ...])

        to wait until the computations of fiber1, fiber2, etc are done.
        :param l: the list of IO to run in parallel.
        :return: the same list where each IO has been replaced by its Fiber
        """
        if len(others) == 1 and isinstance(others[0], abc.Iterable):
            return IO(IOTag.PARALLEL, list((self, *others[0])))
        return IO(IOTag.PARALLEL, list((self, *others)))

    def flatten(self):
        """
        Concatenation function on IO
        """
        if self.__tag == 0:
            return self.__fields
        return IO(IOTag.FLATTEN, self)

    # Reader API

    def contra_map_read(self, f: Callable[[R2], R]) -> IO[R2, E, A]:
        """
        Transform the context with f.
        Note that f is not from R to R2 but from R2 to R!
        """
        return IO(IOTag.CONTRA_MAP_READ, (f, self))

    # Errors API

    def catch(self, handler: Callable[[List[E]], IO[R, E, A]]) -> IO[R, E, A]:
        """
        React to errors (the except part of a try-except).

        On errors, call the handler with the errors.
        """
        return IO(IOTag.CATCH, (self, handler))

    def map_error(self, f: Callable[[E], E2]) -> IO[R, E2, A]:
        """
        Transform the stored errors if the computation fails on an errors.
        Do nothing otherwise.
        """
        return IO(IOTag.MAP_ERROR, (self, f))

    # Panic

    def recover(
        self, handler: Callable[[List[TracedException], List[E]], IO[R, E, A]]
    ) -> IO[R, E, A]:
        """
        React to panics (the except part of a try-except).

        On panic, call the handler with the exceptions.
        """
        return IO(IOTag.RECOVER, (self, handler))

    def map_panic(self, f: Callable[[TracedException], TracedException]) -> IO[R, E, A]:
        """
        Transform the exceptions stored if the computation fails on a panic.
        Do nothing otherwise.
        """
        return IO(IOTag.MAP_PANIC, (self, f))

    def run(self, context: R, pool_size: int = 1, nighttime=0.01) -> Result[E, A]:
        """
        Run the computation.

        Note that a IO is a data structure, no action is performed until you
        call run. You may view an IO value as a function declaration.
        Declaring a function does not execute its body. Only calling the
        function does. Likewise, declaring an IO does not execute its content,
        only running the IO does.

        Note that the return value is a  `Result[E,A]`.
        No exceptions will be raised by run (unless there is a bug), run will
        returns a panic instead!
        """
        from raffiot._runtime import SharedState

        return SharedState(pool_size, nighttime).run(self, context)

    def ap(self: IO[R, E, Callable[[X], A]], *arg: IO[R, E, X]) -> IO[R, E, A]:
        """
        Noting functions from [X1,...,XN] to A: `[X1, ..., Xn] -> A`.

        If self computes a function `f: [X1,...,XN] -> A`
        and arg computes a value `x1: X1`,...,`xn: Xn`
        then self.ap(arg) computes `f(x1,...,xn): A`.
        """
        return self.zip(*arg).map(lambda l: l[0](*l[1:]))  # type: ignore

    def attempt(self) -> IO[R, E, Result[E, A]]:
        """
        Transform this computation that may fail into a computation
        that never fails but returns a Result[E,A].

        - If `self` successfully computes a, then `self.attempt()` successfully computes `Ok(a)`.
        - If `self` fails on errors e, then `self.attempt()` successfully computes `Errors(e)`.
        - If `self` fails on traced exceptions p and errors e, then `self.attempt()` successfully computes `Panic(p,e)`.

        Note that errors and panics stop the computation, unless a catch or
        recover reacts to such failures. But using map, flat_map, flatten and
        ap is sometimes easier than using catch and recover. attempt transforms
        a failed computation into a successful computation returning a failure,
        thus enabling you to use map, flat_map, ... to deal with errors.
        """
        return IO(IOTag.ATTEMPT, self)

    def finally_(self, after: Callable[[Result[E, A]], IO[R, E, Any]]) -> IO[R, E, A]:
        """
        After having computed self, but before returning its result,
        execute the io computation.

        This is extremely useful when you need to perform an action,
        unconditionally, at the end of a computation, without changing
        its result, like releasing a resource.
        """
        return self.attempt().flat_map(
            lambda r1: after(r1)
            .attempt()
            .flat_map(lambda r2: from_result(result.sequence(r2, r1)))
        )

    def on_failure(
        self, handler: Callable[[Result[E, Any]], IO[R, E, A]]
    ) -> IO[R, E, A]:
        """
        Combined form of catch and recover.
        React to any failure of the computation.
        Do nothing if the computation is successful.

        - The handler will be called on `Errors(e)` if the computation fails with errors e.
        - The handler will be called on `Panic(p,e)` if the computation fails with panic p and errors e.
        - The handler will never be called on `Ok(a)`.
        """

        def g(r: Result[E, A]) -> IO[R, E, A]:
            if isinstance(r, Ok):
                return IO(IOTag.PURE, r.success)
            return handler(r)

        return self.attempt().flat_map(g)

    def then_keep(self, *args: IO[R, E, A]) -> IO[R, E, A]:
        """
        Equivalent to `then(*args) but, on success, the computed value
        is self's one.

        Used to execute some IO after a successful computation without
        changing its value.
        :param args:
        :return:
        """
        return self.flat_map(lambda a: sequence(args).then(pure(a)))

    def __str__(self) -> str:
        if self.__tag == IOTag.PURE:
            return f"Pure({self.__fields})"
        if self.__tag == IOTag.MAP:
            return f"Map({self.__fields})"
        if self.__tag == IOTag.FLATMAP:
            return f"FlatMap({self.__fields})"
        if self.__tag == IOTag.FLATTEN:
            return f"Flatten({self.__fields})"
        if self.__tag == IOTag.SEQUENCE:
            return f"Sequence({self.__fields})"
        if self.__tag == IOTag.ZIP:
            return f"Zip({self.__fields})"
        if self.__tag == IOTag.DEFER:
            return f"Defer({self.__fields})"
        if self.__tag == IOTag.DEFER_IO:
            return f"DeferIO({self.__fields})"
        if self.__tag == IOTag.ATTEMPT:
            return f"Attempt({self.__fields})"
        if self.__tag == IOTag.READ:
            return f"Read({self.__fields})"
        if self.__tag == IOTag.CONTRA_MAP_READ:
            return f"ContraMapRead({self.__fields})"
        if self.__tag == IOTag.ERRORS:
            return f"Errors({self.__fields})"
        if self.__tag == IOTag.CATCH:
            return f"Catch({self.__fields})"
        if self.__tag == IOTag.MAP_ERROR:
            return f"MapError({self.__fields})"
        if self.__tag == IOTag.PANIC:
            return f"Panic({self.__fields})"
        if self.__tag == IOTag.RECOVER:
            return f"Recover({self.__fields})"
        if self.__tag == IOTag.MAP_PANIC:
            return f"MapPanic({self.__fields})"
        if self.__tag == IOTag.YIELD:
            return f"Yield({self.__fields})"
        if self.__tag == IOTag.ASYNC:
            return f"Async({self.__fields})"
        if self.__tag == IOTag.DEFER_READ:
            return f"DeferRead({self.__fields})"
        if self.__tag == IOTag.DEFER_READ_IO:
            return f"DeferReadIO({self.__fields})"
        if self.__tag == IOTag.PARALLEL:
            return f"Parallel({self.__fields})"
        if self.__tag == IOTag.WAIT:
            return f"Wait({self.__fields})"
        if self.__tag == IOTag.SLEEP_UNTIL:
            return f"SleepUntil({self.__fields})"
        if self.__tag == IOTag.REC:
            return f"Rec({self.__fields})"
        if self.__tag == IOTag.ACQUIRE:
            return f"Acquire({self.__fields})"
        if self.__tag == IOTag.RELEASE:
            return f"Release({self.__fields})"
        raise MatchError(f"{self} should be an IO")

    def __repr__(self):
        return str(self)

Ancestors

• typing.Generic

Methods

def ap(self: IO[R, E, Callable[[X], A]], *arg: IO[R, E, X]) ‑> IO[~R, ~E, ~A]

Noting functions from [X1,…,XN] to A: [X1, ..., Xn] -> A.

If self computes a function f: [X1,...,XN] -> A and arg computes a value x1: X1,…,xn: Xn then self.ap(arg) computes f(x1,...,xn): A.

Expand source code

def ap(self: IO[R, E, Callable[[X], A]], *arg: IO[R, E, X]) -> IO[R, E, A]:
    """
    Noting functions from [X1,...,XN] to A: `[X1, ..., Xn] -> A`.

    If self computes a function `f: [X1,...,XN] -> A`
    and arg computes a value `x1: X1`,...,`xn: Xn`
    then self.ap(arg) computes `f(x1,...,xn): A`.
    """
    return self.zip(*arg).map(lambda l: l[0](*l[1:]))  # type: ignore

def attempt(self) ‑> IO[~R, ~E, Result[~E, ~A]]

Transform this computation that may fail into a computation that never fails but returns a Result[E,A].

• If self successfully computes a, then self.attempt() successfully computes Ok(a).
• If self fails on errors e, then self.attempt() successfully computes Errors(e).
• If self fails on traced exceptions p and errors e, then self.attempt() successfully computes Panic(p,e).

Note that errors and panics stop the computation, unless a catch or recover reacts to such failures. But using map, flat_map, flatten and ap is sometimes easier than using catch and recover. attempt transforms a failed computation into a successful computation returning a failure, thus enabling you to use map, flat_map, … to deal with errors.

Expand source code

def attempt(self) -> IO[R, E, Result[E, A]]:
    """
    Transform this computation that may fail into a computation
    that never fails but returns a Result[E,A].

    - If `self` successfully computes a, then `self.attempt()` successfully computes `Ok(a)`.
    - If `self` fails on errors e, then `self.attempt()` successfully computes `Errors(e)`.
    - If `self` fails on traced exceptions p and errors e, then `self.attempt()` successfully computes `Panic(p,e)`.

    Note that errors and panics stop the computation, unless a catch or
    recover reacts to such failures. But using map, flat_map, flatten and
    ap is sometimes easier than using catch and recover. attempt transforms
    a failed computation into a successful computation returning a failure,
    thus enabling you to use map, flat_map, ... to deal with errors.
    """
    return IO(IOTag.ATTEMPT, self)

def catch(self, handler: Callable[[List[E]], IO[R, E, A]]) ‑> IO[~R, ~E, ~A]

React to errors (the except part of a try-except).

On errors, call the handler with the errors.

Expand source code

def catch(self, handler: Callable[[List[E]], IO[R, E, A]]) -> IO[R, E, A]:
    """
    React to errors (the except part of a try-except).

    On errors, call the handler with the errors.
    """
    return IO(IOTag.CATCH, (self, handler))

def contra_map_read(self, f: Callable[[R2], R]) ‑> IO[~R2, ~E, ~A]

Transform the context with f. Note that f is not from R to R2 but from R2 to R!

Expand source code

def contra_map_read(self, f: Callable[[R2], R]) -> IO[R2, E, A]:
    """
    Transform the context with f.
    Note that f is not from R to R2 but from R2 to R!
    """
    return IO(IOTag.CONTRA_MAP_READ, (f, self))

def finally_(self, after: Callable[[Result[E, A]], IO[R, E, Any]]) ‑> IO[~R, ~E, ~A]

After having computed self, but before returning its result, execute the io computation.

This is extremely useful when you need to perform an action, unconditionally, at the end of a computation, without changing its result, like releasing a resource.

Expand source code

def finally_(self, after: Callable[[Result[E, A]], IO[R, E, Any]]) -> IO[R, E, A]:
    """
    After having computed self, but before returning its result,
    execute the io computation.

    This is extremely useful when you need to perform an action,
    unconditionally, at the end of a computation, without changing
    its result, like releasing a resource.
    """
    return self.attempt().flat_map(
        lambda r1: after(r1)
        .attempt()
        .flat_map(lambda r2: from_result(result.sequence(r2, r1)))
    )

def flat_map(self, f: Callable[[A], IO[R, E, A2]]) ‑> IO[~R, ~E, ~A2]

Chain two computations. The result of the first one (self) can be used in the second (f).

Expand source code

def flat_map(self, f: Callable[[A], IO[R, E, A2]]) -> IO[R, E, A2]:
    """
    Chain two computations.
    The result of the first one (self) can be used in the second (f).
    """
    return IO(IOTag.FLATMAP, (self, f))

def flatten(self)

Concatenation function on IO

Expand source code

def flatten(self):
    """
    Concatenation function on IO
    """
    if self.__tag == 0:
        return self.__fields
    return IO(IOTag.FLATTEN, self)

def map(self, f: Callable[[A], A2]) ‑> IO[~R, ~E, ~A2]

Transform the computed value with f if the computation is successful. Do nothing otherwise.

Expand source code

def map(self, f: Callable[[A], A2]) -> IO[R, E, A2]:
    """
    Transform the computed value with f if the computation is successful.
    Do nothing otherwise.
    """
    return IO(IOTag.MAP, (self, f))

def map_error(self, f: Callable[[E], E2]) ‑> IO[~R, ~E2, ~A]

Transform the stored errors if the computation fails on an errors. Do nothing otherwise.

Expand source code

def map_error(self, f: Callable[[E], E2]) -> IO[R, E2, A]:
    """
    Transform the stored errors if the computation fails on an errors.
    Do nothing otherwise.
    """
    return IO(IOTag.MAP_ERROR, (self, f))

def map_panic(self, f: Callable[[TracedException], TracedException]) ‑> IO[~R, ~E, ~A]

Transform the exceptions stored if the computation fails on a panic. Do nothing otherwise.

Expand source code

def map_panic(self, f: Callable[[TracedException], TracedException]) -> IO[R, E, A]:
    """
    Transform the exceptions stored if the computation fails on a panic.
    Do nothing otherwise.
    """
    return IO(IOTag.MAP_PANIC, (self, f))

def on_failure(self, handler: Callable[[Result[E, Any]], IO[R, E, A]]) ‑> IO[~R, ~E, ~A]

Combined form of catch and recover. React to any failure of the computation. Do nothing if the computation is successful.

• The handler will be called on Errors(e) if the computation fails with errors e.
• The handler will be called on Panic(p,e) if the computation fails with panic p and errors e.
• The handler will never be called on Ok(a).

Expand source code

def on_failure(
    self, handler: Callable[[Result[E, Any]], IO[R, E, A]]
) -> IO[R, E, A]:
    """
    Combined form of catch and recover.
    React to any failure of the computation.
    Do nothing if the computation is successful.

    - The handler will be called on `Errors(e)` if the computation fails with errors e.
    - The handler will be called on `Panic(p,e)` if the computation fails with panic p and errors e.
    - The handler will never be called on `Ok(a)`.
    """

    def g(r: Result[E, A]) -> IO[R, E, A]:
        if isinstance(r, Ok):
            return IO(IOTag.PURE, r.success)
        return handler(r)

    return self.attempt().flat_map(g)

def parallel(self: IO[R, E, A], *others: IO[R, E, A]) ‑> IO[~R, ~E, typing.List[Fiber]]

Run all these IO (including self) in parallel. Return the list of fibers, in the same order.

Each Fiber represent a parallel computation. Call

>>> wait([fiber1, fiber2, ...])

to wait until the computations of fiber1, fiber2, etc are done. :param l: the list of IO to run in parallel. :return: the same list where each IO has been replaced by its Fiber

Expand source code

def parallel(self: IO[R, E, A], *others: IO[R, E, A]) -> IO[R, E, List[Fiber]]:
    """
    Run all these IO (including self) in parallel.
    Return the list of fibers, in the same order.

    Each Fiber represent a parallel computation. Call

    >>> wait([fiber1, fiber2, ...])

    to wait until the computations of fiber1, fiber2, etc are done.
    :param l: the list of IO to run in parallel.
    :return: the same list where each IO has been replaced by its Fiber
    """
    if len(others) == 1 and isinstance(others[0], abc.Iterable):
        return IO(IOTag.PARALLEL, list((self, *others[0])))
    return IO(IOTag.PARALLEL, list((self, *others)))

def recover(self, handler: Callable[[List[TracedException], List[E]], IO[R, E, A]]) ‑> IO[~R, ~E, ~A]

React to panics (the except part of a try-except).

On panic, call the handler with the exceptions.

Expand source code

def recover(
    self, handler: Callable[[List[TracedException], List[E]], IO[R, E, A]]
) -> IO[R, E, A]:
    """
    React to panics (the except part of a try-except).

    On panic, call the handler with the exceptions.
    """
    return IO(IOTag.RECOVER, (self, handler))

def run(self, context: R, pool_size: int = 1, nighttime=0.01) ‑> Result[~E, ~A]

Run the computation.

Note that a IO is a data structure, no action is performed until you call run. You may view an IO value as a function declaration. Declaring a function does not execute its body. Only calling the function does. Likewise, declaring an IO does not execute its content, only running the IO does.

Note that the return value is a Result[E,A]. No exceptions will be raised by run (unless there is a bug), run will returns a panic instead!

Expand source code

def run(self, context: R, pool_size: int = 1, nighttime=0.01) -> Result[E, A]:
    """
    Run the computation.

    Note that a IO is a data structure, no action is performed until you
    call run. You may view an IO value as a function declaration.
    Declaring a function does not execute its body. Only calling the
    function does. Likewise, declaring an IO does not execute its content,
    only running the IO does.

    Note that the return value is a  `Result[E,A]`.
    No exceptions will be raised by run (unless there is a bug), run will
    returns a panic instead!
    """
    from raffiot._runtime import SharedState

    return SharedState(pool_size, nighttime).run(self, context)

def then(self, *others: IO[R, E, A2]) ‑> IO[~R, ~E, ~A2]

Chain two computations. The result of the first one (self) is dropped.

Expand source code

def then(self, *others: IO[R, E, A2]) -> IO[R, E, A2]:
    """
    Chain two computations.
    The result of the first one (self) is dropped.
    """
    if len(others) == 1 and isinstance(others[0], abc.Iterable):
        return IO(IOTag.SEQUENCE, list((self, *others[0])))
    return IO(IOTag.SEQUENCE, list((self, *others)))

def then_keep(self, *args: IO[R, E, A]) ‑> IO[~R, ~E, ~A]

Equivalent to `then(*args) but, on success, the computed value is self's one.

Used to execute some IO after a successful computation without changing its value. :param args: :return:

Expand source code

def then_keep(self, *args: IO[R, E, A]) -> IO[R, E, A]:
    """
    Equivalent to `then(*args) but, on success, the computed value
    is self's one.

    Used to execute some IO after a successful computation without
    changing its value.
    :param args:
    :return:
    """
    return self.flat_map(lambda a: sequence(args).then(pure(a)))

def zip(self: IO[R, E, A], *others: IO[R, E, X]) ‑> IO[~R, ~E, typing.Iterable[~A]]

Pack a list of IO (including self) into an IO computing the list of all values.

If one IO fails, the whole computation fails.

Expand source code

def zip(self: IO[R, E, A], *others: IO[R, E, X]) -> IO[R, E, Iterable[A]]:
    """
    Pack a list of IO (including self) into an IO computing the list
    of all values.

    If one IO fails, the whole computation fails.
    """
    if len(others) == 1 and isinstance(others[0], abc.Iterable):
        return IO(IOTag.ZIP, list((self, *others[0])))
    return IO(IOTag.ZIP, list((self, *others)))

def zip_par(self: IO[R, E, A], *others: IO[R, E, A]) ‑> IO[~R, ~E, typing.List[~A]]

Pack a list of IO (including self) into an IO computing the list of all values in parallel.

If one IO fails, the whole computation fails.

Expand source code

def zip_par(self: IO[R, E, A], *others: IO[R, E, A]) -> IO[R, E, List[A]]:
    """
    Pack a list of IO (including self) into an IO computing the list
    of all values in parallel.

    If one IO fails, the whole computation fails.
    """
    return zip_par(self, *others)

@dataclass
class MatchError(Exception):
    """
    Exception for pattern matching errors (used internally, should NEVER happen).
    """

    message: str

Ancestors

• builtins.Exception
• builtins.BaseException

Class variables

var message : str

@final
@dataclass
class MultipleExceptions(Exception, Generic[E]):
    """
    Represents
    """

    exceptions: List[TracedException]
    """
    The list exceptions encountered
    """

    errors: List[E]
    """
    The list of errors encountered
    """

    @classmethod
    def merge(cls, *exceptions: TracedException, errors: List[E] = None) -> Exception:
        """
        Merge some exceptions, retuning the exceptions if there is only one
        or a  `MultipleExceptions` otherwise.

        :param exceptions:
        :param errors:
        :return:
        """
        stack = [exn for exn in exceptions]
        base_exceptions = []
        errs = [x for x in errors] if errors else []

        while stack:
            item = stack.pop()
            if isinstance(item, MultipleExceptions):
                stack.extend(item.exceptions)
                errs.extend(item.errors)
                continue
            if isinstance(item, abc.Iterable) and not isinstance(item, str):
                stack.extend(item)
                continue
            base_exceptions.append(TracedException.ensure_traced(item))

        base_exceptions.reverse()
        return MultipleExceptions(base_exceptions, errs)

    def __str__(self):
        msg = ""
        for traced in self.exceptions:
            msg += f"\nException: {traced.exception}\n{traced.stack_trace}"
        for err in self.errors:
            msg += f"\nError: {err}"
        return msg

Ancestors

• builtins.Exception
• builtins.BaseException
• typing.Generic

Class variables

var errors : List[+E]

The list of errors encountered

var exceptions : List[TracedException]

The list exceptions encountered

Static methods

def merge(*exceptions: TracedException, errors: List[E] = None) ‑> Exception

Merge some exceptions, retuning the exceptions if there is only one or a MultipleExceptions otherwise.

:param exceptions: :param errors: :return:

Expand source code

@classmethod
def merge(cls, *exceptions: TracedException, errors: List[E] = None) -> Exception:
    """
    Merge some exceptions, retuning the exceptions if there is only one
    or a  `MultipleExceptions` otherwise.

    :param exceptions:
    :param errors:
    :return:
    """
    stack = [exn for exn in exceptions]
    base_exceptions = []
    errs = [x for x in errors] if errors else []

    while stack:
        item = stack.pop()
        if isinstance(item, MultipleExceptions):
            stack.extend(item.exceptions)
            errs.extend(item.errors)
            continue
        if isinstance(item, abc.Iterable) and not isinstance(item, str):
            stack.extend(item)
            continue
        base_exceptions.append(TracedException.ensure_traced(item))

    base_exceptions.reverse()
    return MultipleExceptions(base_exceptions, errs)

@final
@dataclass
class Ok(Result[E, A]):
    """
    The result of a successful computation.
    """

    __slots__ = "success"

    success: A

Ancestors

• Result
• typing.Generic

Instance variables

var success : ~A

Return an attribute of instance, which is of type owner.

Inherited members

• Result:
  • flatten
  • is_error
  • is_ok
  • is_panic
  • raise_on_panic
  • to_computation_status
  • unsafe_ap
  • unsafe_catch
  • unsafe_flat_map
  • unsafe_fold
  • unsafe_fold_raise
  • unsafe_get
  • unsafe_map
  • unsafe_map_error
  • unsafe_map_panic
  • unsafe_recover
  • unsafe_tri_map
  • zip

@final
@dataclass
class Panic(Result[E, A]):
    """
    The result of a computation that failed unexpectedly.
    The program is not in a valid state and must terminate safely.
    """

    __slots__ = "exceptions", "errors"

    exceptions: List[TracedException]

    errors: List[E]

Ancestors

• Result
• typing.Generic

Instance variables

var errors : List[~E]

Return an attribute of instance, which is of type owner.

var exceptions : List[TracedException]

Return an attribute of instance, which is of type owner.

Inherited members

• Result:
  • flatten
  • is_error
  • is_ok
  • is_panic
  • raise_on_panic
  • to_computation_status
  • unsafe_ap
  • unsafe_catch
  • unsafe_flat_map
  • unsafe_fold
  • unsafe_fold_raise
  • unsafe_get
  • unsafe_map
  • unsafe_map_error
  • unsafe_map_panic
  • unsafe_recover
  • unsafe_tri_map
  • zip

@final
@dataclass
class Resource(Generic[R, E, A]):
    """
    Essentially an IO-powered data structure that produces resources of type A,
    can fail with errors of type E and read a context of type R.
    """

    __slots__ = "create"

    create: IO[R, E, Tuple[A, Callable[[ComputationStatus], IO[R, E, Any]]]]
    """
    IO to create one resource along with the IO for releasing it.
    
    On success, this IO must produce a `Tuple[A, IO[R,Any,Any]`:
        - The first value of the tuple, of type `A`, is the created resource.
        - The second value of the tuple, of type `Callable[[ComputationStatus], IO[R,E,Any]]`,
          is the function that releases the resource. 
          It receives the `Result` (with the value set to None) to indicate
          whether the computation succeeded, failed or panicked.
    
    For example:
        
        >>> Resource(io.defer(lambda: open("file")).map(
        >>>     lambda a: (a, lambda _:io.defer(a.close))))
    """

    def use(self, fun: Callable[[A], IO[R, E, X]]) -> IO[R, E, X]:
        """
        Create a resource a:A and use it in fun.

        Once created, the resource a:A is guaranteed to be released (by running
        its releasing IO). The return value if the result of fun(a).

        This is the only way to use a resource.
        """

        def safe_use(
            x: Tuple[A, Callable[[ComputationStatus], IO[R, E, None]]]
        ) -> IO[R, E, X]:
            a, close = x
            return io.defer_io(fun, a).finally_(
                lambda r: close(r.to_computation_status())
            )

        return self.create.flat_map(safe_use)

    def with_(self, the_io: IO[R, E, X]) -> IO[R, E, X]:
        """
        Create a resource a:A but does not use it in the IO.

        Once created, the resource a:A is guaranteed to be released (by running
        its releasing IO). The return value if the result of the_io.

        This is an alias for self.use(lambda _: the_io)
        """
        return self.use(lambda _: the_io)

    def map(self, f: Callable[[A], A2]) -> Resource[R, E, A2]:
        """
        Transform the created resource with f if the creation is successful.
        Do nothing otherwise.
        """

        def safe_map(
            x: Tuple[A, Callable[[ComputationStatus], IO[R, E, None]]]
        ) -> IO[R, E, Tuple[A2, Callable[[ComputationStatus], IO[R, E, None]]]]:
            a, close = x
            try:
                return io.pure((f(a), close))
            except Exception as exception:
                return close_and_merge_failure(
                    close,
                    Panic(
                        exceptions=[TracedException.in_except_clause(exception)],
                        errors=[],
                    ),
                )

        return Resource(self.create.flat_map(safe_map))

    def flat_map(self, f: Callable[[A], Resource[R, E, A2]]) -> Resource[R, E, A2]:
        """
        Chain two Resource.
        The resource created by the first one (self) can be used to create the second (f).
        """

        def safe_flat_map_a(
            xa: Tuple[A, Callable[[ComputationStatus], IO[R, E, None]]]
        ) -> IO[R, E, Tuple[A2, Callable[[ComputationStatus], IO[R, E, None]]]]:
            a, close_a = xa

            def safe_flat_map_a2(
                xa2: Tuple[A2, Callable[[ComputationStatus], IO[R, Any, None]]]
            ) -> IO[R, E, Tuple[A2, Callable[[ComputationStatus], IO[R, Any, None]]]]:
                a2, close_a2 = xa2

                def close_all(cs: ComputationStatus) -> IO[R, E, Any]:
                    return io.zip(
                        io.defer_io(close_a2, cs).attempt(),
                        io.defer_io(close_a, cs).attempt(),
                    ).flat_map(lambda rs: io.from_result(result.zip(rs)))

                return io.pure((a2, close_all))

            return (
                io.defer_io(lambda x: f(x).create, a)
                .attempt()
                .flat_map(
                    lambda r: r.unsafe_fold(
                        safe_flat_map_a2,
                        lambda e: close_and_merge_failure(close_a, Errors(e)),
                        lambda p, e: close_and_merge_failure(close_a, Panic(p, e)),
                    )
                )
            )

        return Resource(self.create.flat_map(safe_flat_map_a))

    def then(self, rs: Resource[R, E, A2]) -> Resource[R, E, A2]:
        """
        Chain two Resource.
        The resource created by the first one (self) is dropped.
        """
        return self.flat_map(lambda _: rs)

    def zip(self: Resource[R, E, A], *rs: Resource[R, E, A]) -> Resource[R, E, List[A]]:
        """
        Pack a list of resources (including self) into a Resource creating the
        list of all resources.

        If one resource creation fails, the whole creation fails (opened resources
        are released then).
        """
        return zip(self, *rs)

    def zip_par(
        self: Resource[R, E, A], *rs: Resource[R, E, A]
    ) -> Resource[R, E, List[A]]:
        """
        Pack a list of resources (including self) into a Resource creating the
        list of all resources.

        If one resource creation fails, the whole creation fails (opened resources
        are released then).
        """
        return zip_par(self, *rs)

    def ap(
        self: Resource[R, E, Callable[[X], A]], *arg: Resource[R, E, X]
    ) -> Resource[R, E, A]:
        """
        Noting functions from [X1,...,XN] to A: `[X1, ..., Xn] -> A`.

        If self computes a function `f: [X1,...,XN] -> A`
        and arg computes a value `x1: X1`,...,`xn: Xn`
        then self.ap(arg) computes `f(x1,...,xn): A`.
        """
        return self.zip(*arg).map(lambda l: l[0](*l[1:]))  # type: ignore

    def flatten(self: Resource[R, E, Resource[R, E, A]]) -> Resource[R, E, A]:
        """
        Concatenation function on Resource
        """
        return self.flat_map(lambda x: x)

    # Reader API

    def contra_map_read(self, f: Callable[[R2], R]) -> Resource[R2, E, A]:
        """
        Transform the context with f.
        Note that f is not from R to R2 but from R2 to R!
        """
        return Resource(
            self.create.contra_map_read(f).map(
                lambda x: (x[0], lambda cs: io.defer_io(x[1], cs).contra_map_read(f))
            )
        )

    # Errors API

    def catch(
        self, handler: Callable[[List[E]], Resource[R, E, A]]
    ) -> Resource[R, E, A]:
        """
        React to errors (the except part of a try-except).

        On errors, call the handler with the errors.
        """
        return Resource(self.create.catch(lambda e: handler(e).create))

    def map_error(self, f: Callable[[E], E2]) -> Resource[R, E2, A]:
        """
        Transform the stored errors if the resource creation fails on an errors.
        Do nothing otherwise.
        """
        return Resource(
            self.create.map_error(f).map(
                lambda x: (x[0], lambda cs: io.defer_io(x[1], cs).map_error(f))
            )
        )

    # Panic

    def recover(
        self, handler: Callable[[List[TracedException], List[E]], Resource[R, E, A]]
    ) -> Resource[R, E, A]:
        """
        React to panics (the except part of a try-except).

        On panic, call the handler with the exceptions.
        """
        return Resource(self.create.recover(lambda p, e: handler(p, e).create))

    def map_panic(
        self, f: Callable[[TracedException], TracedException]
    ) -> Resource[R, E, A]:
        """
        Transform the exceptions stored if the computation fails on a panic.
        Do nothing otherwise.
        """
        return Resource(
            self.create.map_panic(f).map(
                lambda x: (x[0], lambda cs: io.defer_io(x[1], cs).map_panic(f))
            )
        )

    def attempt(self) -> Resource[R, Any, Result[E, A]]:
        """
        Transform this Resource that may fail into a Resource
        that never fails but creates a Result[E,A].

        - If self successfully computes a, then `self.attempt()` successfully computes `Ok(a)`.
        - If self fails on errors e, then `self.attempt()` successfully computes `Errors(e)`.
        - If self fails on panic p, then `self.attempt()` successfully computes `Panic(p)`.

        Note that errors and panics stop the resource creation, unless a catch or
        recover reacts to such failures. But using map, flat_map, flatten and
        ap is sometimes easier than using catch and recover. attempt transforms
        a failed resource creation into a successful resource creation returning a failure,
        thus enabling you to use map, flat_map, ... to deal with errors.
        """
        return Resource(
            self.create.attempt().map(
                lambda x: x.unsafe_fold(
                    lambda v: (Ok(v[0]), v[1]),
                    lambda e: (Errors(e), noop_close),
                    lambda p, e: (Panic(p, e), noop_close),
                )
            )
        )

    def finally_(
        self, after: Callable[[Result[E, A]], Resource[R, E, Any]]
    ) -> Resource[R, E, A]:
        """
        After having computed self, but before returning its result,
        execute the rs Resource creation.

        This is extremely useful when you need to perform an action,
        unconditionally, at the end of a resource creation, without changing
        its result, like executing a lifted IO.
        """
        return self.attempt().flat_map(
            lambda r1: after(r1)
            .attempt()
            .flat_map(lambda r2: from_result(result.sequence(r2, r1)))
        )

    def on_failure(
        self, handler: Callable[[Result[E, Any]], Resource[R, E, A]]
    ) -> Resource[R, E, A]:
        """
        Combined form of catch and recover.
        React to any failure of the resource creation.
        Do nothing if the resource creation is successful.

        - The handler will be called on `Errors(e)` if the resource creation fails with errors e.
        - The handler will be called on `Panic(p)` if the resource creation fails with panic p.
        - The handler will never be called on `Ok(a)`.
        """

        def g(r: Result[E, A]) -> Resource[R, E, A]:
            if isinstance(r, Ok):
                return pure(r.success)
            return handler(r)

        return self.attempt().flat_map(g)

Ancestors

• typing.Generic

Instance variables

var create : IO[~R, ~E, typing.Tuple[~A, typing.Callable[[ComputationStatus], IO[~R, ~E, typing.Any]]]]

IO to create one resource along with the IO for releasing it.

On success, this IO must produce a Tuple[A, IO[R,Any,Any]: - The first value of the tuple, of type A, is the created resource. - The second value of the tuple, of type Callable[[ComputationStatus], IO[R,E,Any]], is the function that releases the resource. It receives the Result (with the value set to None) to indicate whether the computation succeeded, failed or panicked.

For example:

>>> Resource(io.defer(lambda: open("file")).map(
>>>     lambda a: (a, lambda _:io.defer(a.close))))

Methods

def ap(self: Resource[R, E, Callable[[X], A]], *arg: Resource[R, E, X]) ‑> Resource[~R, ~E, ~A]

Noting functions from [X1,…,XN] to A: [X1, ..., Xn] -> A.

If self computes a function f: [X1,...,XN] -> A and arg computes a value x1: X1,…,xn: Xn then self.ap(arg) computes f(x1,...,xn): A.

Expand source code

def ap(
    self: Resource[R, E, Callable[[X], A]], *arg: Resource[R, E, X]
) -> Resource[R, E, A]:
    """
    Noting functions from [X1,...,XN] to A: `[X1, ..., Xn] -> A`.

    If self computes a function `f: [X1,...,XN] -> A`
    and arg computes a value `x1: X1`,...,`xn: Xn`
    then self.ap(arg) computes `f(x1,...,xn): A`.
    """
    return self.zip(*arg).map(lambda l: l[0](*l[1:]))  # type: ignore

def attempt(self) ‑> Resource[~R, typing.Any, Result[~E, ~A]]

Transform this Resource that may fail into a Resource that never fails but creates a Result[E,A].

• If self successfully computes a, then self.attempt() successfully computes Ok(a).
• If self fails on errors e, then self.attempt() successfully computes Errors(e).
• If self fails on panic p, then self.attempt() successfully computes Panic(p).

Note that errors and panics stop the resource creation, unless a catch or recover reacts to such failures. But using map, flat_map, flatten and ap is sometimes easier than using catch and recover. attempt transforms a failed resource creation into a successful resource creation returning a failure, thus enabling you to use map, flat_map, … to deal with errors.

Expand source code

def attempt(self) -> Resource[R, Any, Result[E, A]]:
    """
    Transform this Resource that may fail into a Resource
    that never fails but creates a Result[E,A].

    - If self successfully computes a, then `self.attempt()` successfully computes `Ok(a)`.
    - If self fails on errors e, then `self.attempt()` successfully computes `Errors(e)`.
    - If self fails on panic p, then `self.attempt()` successfully computes `Panic(p)`.

    Note that errors and panics stop the resource creation, unless a catch or
    recover reacts to such failures. But using map, flat_map, flatten and
    ap is sometimes easier than using catch and recover. attempt transforms
    a failed resource creation into a successful resource creation returning a failure,
    thus enabling you to use map, flat_map, ... to deal with errors.
    """
    return Resource(
        self.create.attempt().map(
            lambda x: x.unsafe_fold(
                lambda v: (Ok(v[0]), v[1]),
                lambda e: (Errors(e), noop_close),
                lambda p, e: (Panic(p, e), noop_close),
            )
        )
    )

def catch(self, handler: Callable[[List[E]], Resource[R, E, A]]) ‑> Resource[~R, ~E, ~A]

React to errors (the except part of a try-except).

On errors, call the handler with the errors.

Expand source code

def catch(
    self, handler: Callable[[List[E]], Resource[R, E, A]]
) -> Resource[R, E, A]:
    """
    React to errors (the except part of a try-except).

    On errors, call the handler with the errors.
    """
    return Resource(self.create.catch(lambda e: handler(e).create))

def contra_map_read(self, f: Callable[[R2], R]) ‑> Resource[~R2, ~E, ~A]

Transform the context with f. Note that f is not from R to R2 but from R2 to R!

Expand source code

def contra_map_read(self, f: Callable[[R2], R]) -> Resource[R2, E, A]:
    """
    Transform the context with f.
    Note that f is not from R to R2 but from R2 to R!
    """
    return Resource(
        self.create.contra_map_read(f).map(
            lambda x: (x[0], lambda cs: io.defer_io(x[1], cs).contra_map_read(f))
        )
    )

def finally_(self, after: Callable[[Result[E, A]], Resource[R, E, Any]]) ‑> Resource[~R, ~E, ~A]

After having computed self, but before returning its result, execute the rs Resource creation.

This is extremely useful when you need to perform an action, unconditionally, at the end of a resource creation, without changing its result, like executing a lifted IO.

Expand source code

def finally_(
    self, after: Callable[[Result[E, A]], Resource[R, E, Any]]
) -> Resource[R, E, A]:
    """
    After having computed self, but before returning its result,
    execute the rs Resource creation.

    This is extremely useful when you need to perform an action,
    unconditionally, at the end of a resource creation, without changing
    its result, like executing a lifted IO.
    """
    return self.attempt().flat_map(
        lambda r1: after(r1)
        .attempt()
        .flat_map(lambda r2: from_result(result.sequence(r2, r1)))
    )

def flat_map(self, f: Callable[[A], Resource[R, E, A2]]) ‑> Resource[~R, ~E, ~A2]

Chain two Resource. The resource created by the first one (self) can be used to create the second (f).

Expand source code

def flat_map(self, f: Callable[[A], Resource[R, E, A2]]) -> Resource[R, E, A2]:
    """
    Chain two Resource.
    The resource created by the first one (self) can be used to create the second (f).
    """

    def safe_flat_map_a(
        xa: Tuple[A, Callable[[ComputationStatus], IO[R, E, None]]]
    ) -> IO[R, E, Tuple[A2, Callable[[ComputationStatus], IO[R, E, None]]]]:
        a, close_a = xa

        def safe_flat_map_a2(
            xa2: Tuple[A2, Callable[[ComputationStatus], IO[R, Any, None]]]
        ) -> IO[R, E, Tuple[A2, Callable[[ComputationStatus], IO[R, Any, None]]]]:
            a2, close_a2 = xa2

            def close_all(cs: ComputationStatus) -> IO[R, E, Any]:
                return io.zip(
                    io.defer_io(close_a2, cs).attempt(),
                    io.defer_io(close_a, cs).attempt(),
                ).flat_map(lambda rs: io.from_result(result.zip(rs)))

            return io.pure((a2, close_all))

        return (
            io.defer_io(lambda x: f(x).create, a)
            .attempt()
            .flat_map(
                lambda r: r.unsafe_fold(
                    safe_flat_map_a2,
                    lambda e: close_and_merge_failure(close_a, Errors(e)),
                    lambda p, e: close_and_merge_failure(close_a, Panic(p, e)),
                )
            )
        )

    return Resource(self.create.flat_map(safe_flat_map_a))

def flatten(self: Resource[R, E, Resource[R, E, A]]) ‑> Resource[~R, ~E, ~A]

Concatenation function on Resource

Expand source code

def flatten(self: Resource[R, E, Resource[R, E, A]]) -> Resource[R, E, A]:
    """
    Concatenation function on Resource
    """
    return self.flat_map(lambda x: x)

def map(self, f: Callable[[A], A2]) ‑> Resource[~R, ~E, ~A2]

Transform the created resource with f if the creation is successful. Do nothing otherwise.

Expand source code

def map(self, f: Callable[[A], A2]) -> Resource[R, E, A2]:
    """
    Transform the created resource with f if the creation is successful.
    Do nothing otherwise.
    """

    def safe_map(
        x: Tuple[A, Callable[[ComputationStatus], IO[R, E, None]]]
    ) -> IO[R, E, Tuple[A2, Callable[[ComputationStatus], IO[R, E, None]]]]:
        a, close = x
        try:
            return io.pure((f(a), close))
        except Exception as exception:
            return close_and_merge_failure(
                close,
                Panic(
                    exceptions=[TracedException.in_except_clause(exception)],
                    errors=[],
                ),
            )

    return Resource(self.create.flat_map(safe_map))

def map_error(self, f: Callable[[E], E2]) ‑> Resource[~R, ~E2, ~A]

Transform the stored errors if the resource creation fails on an errors. Do nothing otherwise.

Expand source code

def map_error(self, f: Callable[[E], E2]) -> Resource[R, E2, A]:
    """
    Transform the stored errors if the resource creation fails on an errors.
    Do nothing otherwise.
    """
    return Resource(
        self.create.map_error(f).map(
            lambda x: (x[0], lambda cs: io.defer_io(x[1], cs).map_error(f))
        )
    )

def map_panic(self, f: Callable[[TracedException], TracedException]) ‑> Resource[~R, ~E, ~A]

Transform the exceptions stored if the computation fails on a panic. Do nothing otherwise.

Expand source code

def map_panic(
    self, f: Callable[[TracedException], TracedException]
) -> Resource[R, E, A]:
    """
    Transform the exceptions stored if the computation fails on a panic.
    Do nothing otherwise.
    """
    return Resource(
        self.create.map_panic(f).map(
            lambda x: (x[0], lambda cs: io.defer_io(x[1], cs).map_panic(f))
        )
    )

def on_failure(self, handler: Callable[[Result[E, Any]], Resource[R, E, A]]) ‑> Resource[~R, ~E, ~A]

Combined form of catch and recover. React to any failure of the resource creation. Do nothing if the resource creation is successful.

• The handler will be called on Errors(e) if the resource creation fails with errors e.
• The handler will be called on Panic(p) if the resource creation fails with panic p.
• The handler will never be called on Ok(a).

Expand source code

def on_failure(
    self, handler: Callable[[Result[E, Any]], Resource[R, E, A]]
) -> Resource[R, E, A]:
    """
    Combined form of catch and recover.
    React to any failure of the resource creation.
    Do nothing if the resource creation is successful.

    - The handler will be called on `Errors(e)` if the resource creation fails with errors e.
    - The handler will be called on `Panic(p)` if the resource creation fails with panic p.
    - The handler will never be called on `Ok(a)`.
    """

    def g(r: Result[E, A]) -> Resource[R, E, A]:
        if isinstance(r, Ok):
            return pure(r.success)
        return handler(r)

    return self.attempt().flat_map(g)

def recover(self, handler: Callable[[List[TracedException], List[E]], Resource[R, E, A]]) ‑> Resource[~R, ~E, ~A]

React to panics (the except part of a try-except).

On panic, call the handler with the exceptions.

Expand source code

def recover(
    self, handler: Callable[[List[TracedException], List[E]], Resource[R, E, A]]
) -> Resource[R, E, A]:
    """
    React to panics (the except part of a try-except).

    On panic, call the handler with the exceptions.
    """
    return Resource(self.create.recover(lambda p, e: handler(p, e).create))

def then(self, rs: Resource[R, E, A2]) ‑> Resource[~R, ~E, ~A2]

Chain two Resource. The resource created by the first one (self) is dropped.

Expand source code

def then(self, rs: Resource[R, E, A2]) -> Resource[R, E, A2]:
    """
    Chain two Resource.
    The resource created by the first one (self) is dropped.
    """
    return self.flat_map(lambda _: rs)

def use(self, fun: Callable[[A], IO[R, E, X]]) ‑> IO[~R, ~E, ~X]

Create a resource a:A and use it in fun.

Once created, the resource a:A is guaranteed to be released (by running its releasing IO). The return value if the result of fun(a).

This is the only way to use a resource.

Expand source code

def use(self, fun: Callable[[A], IO[R, E, X]]) -> IO[R, E, X]:
    """
    Create a resource a:A and use it in fun.

    Once created, the resource a:A is guaranteed to be released (by running
    its releasing IO). The return value if the result of fun(a).

    This is the only way to use a resource.
    """

    def safe_use(
        x: Tuple[A, Callable[[ComputationStatus], IO[R, E, None]]]
    ) -> IO[R, E, X]:
        a, close = x
        return io.defer_io(fun, a).finally_(
            lambda r: close(r.to_computation_status())
        )

    return self.create.flat_map(safe_use)

def with_(self, the_io: IO[R, E, X]) ‑> IO[~R, ~E, ~X]

Create a resource a:A but does not use it in the IO.

Once created, the resource a:A is guaranteed to be released (by running its releasing IO). The return value if the result of the_io.

This is an alias for self.use(lambda _: the_io)

Expand source code

def with_(self, the_io: IO[R, E, X]) -> IO[R, E, X]:
    """
    Create a resource a:A but does not use it in the IO.

    Once created, the resource a:A is guaranteed to be released (by running
    its releasing IO). The return value if the result of the_io.

    This is an alias for self.use(lambda _: the_io)
    """
    return self.use(lambda _: the_io)

def zip(self: Resource[R, E, A], *rs: Resource[R, E, A]) ‑> Resource[~R, ~E, typing.List[~A]]

Pack a list of resources (including self) into a Resource creating the list of all resources.

If one resource creation fails, the whole creation fails (opened resources are released then).

Expand source code

def zip(self: Resource[R, E, A], *rs: Resource[R, E, A]) -> Resource[R, E, List[A]]:
    """
    Pack a list of resources (including self) into a Resource creating the
    list of all resources.

    If one resource creation fails, the whole creation fails (opened resources
    are released then).
    """
    return zip(self, *rs)

def zip_par(self: Resource[R, E, A], *rs: Resource[R, E, A]) ‑> Resource[~R, ~E, typing.List[~A]]

Pack a list of resources (including self) into a Resource creating the list of all resources.

If one resource creation fails, the whole creation fails (opened resources are released then).

Expand source code

def zip_par(
    self: Resource[R, E, A], *rs: Resource[R, E, A]
) -> Resource[R, E, List[A]]:
    """
    Pack a list of resources (including self) into a Resource creating the
    list of all resources.

    If one resource creation fails, the whole creation fails (opened resources
    are released then).
    """
    return zip_par(self, *rs)

class Result(Generic[E, A]):
    """
    The Result[E,A] data structure represents the result of a computation. It has
    3 possible cases:

    - *Ok(some_value: A)*
            The computation succeeded.
            The value some_value, of type A, is the result of the computation
    - *Errors(some_errors: List[E])*
            The computation failed with an expected errors.
            The errors some_errors, of type List[E], is the expected errors encountered.
    - *Panic(some_exceptions: List[TracedException], errors: List[E])*
            The computation failed on an unexpected errors.
            The exceptions some_exceptions is the unexpected errors encountered.

    The distinction between errors (expected failures) and panics (unexpected
    failures) is essential.

    Errors are failures your program is prepared to deal with safely. An errors
    simply means some operation was not successful, but your program is still
    behaving nicely. Nothing terribly wrong happened. Generally errors belong to
    your business domain. You can take any type as E.

    Panics, on the contrary, are failures you never expected. Your computation
    can not progress further. All you can do when panics occur, is stopping your
    computation gracefully (like releasing resources before dying). The panic type
    is always TracedException.

    As an example, if your program is an HTTP server. Errors are bad requests
    (errors code 400) while panics are internal server errors (errors code 500).
    Receiving bad request is part of the normal life of any HTTP server, it must
    know how to reply nicely. But internal server errors are bugs.
    """

    @final
    def unsafe_fold(
        self,
        on_success: Callable[[A], X],
        on_error: Callable[[List[E]], X],
        on_panic: Callable[[List[TracedException], List[E]], X],
    ) -> X:
        """
        Transform this Result[E,A] into X.
        :param on_success: is called if this result is a `Ok`.
        :param on_error: is called if this result is a `Errors`.
        :param on_panic: is called if this result is a `Panic`.
        :return:
        """
        if isinstance(self, Ok):
            return on_success(self.success)
        if isinstance(self, Errors):
            return on_error(self.errors)
        if isinstance(self, Panic):
            return on_panic(self.exceptions, self.errors)
        raise on_panic([MatchError(f"{self} should be a Result")], [])

    @final
    @safe
    def fold(
        self,
        on_success: Callable[[A], X],
        on_error: Callable[[List[E]], X],
        on_panic: Callable[[List[TracedException], List[E]], X],
    ) -> X:
        """
        Transform this Result[E,A] into X.
        :param on_success: is called if this result is a `Ok`.
        :param on_error: is called if this result is a `Errors`.
        :param on_panic: is called if this result is a `Panic`.
        :return:
        """
        return self.unsafe_fold(on_success, on_error, on_panic)

    @final
    def unsafe_fold_raise(
        self, on_success: Callable[[A], X], on_error: Callable[[List[E]], X]
    ) -> X:
        """
        Transform this `Result[E,A]` into `X` if this result is an `Ok` or `Errors`.
        But raise the stored exceptions is this is a panic.

        It is useful to raise an exceptions on panics.

        :param on_success: is called if this result is a `Ok`.
        :param on_error: is called if this result is a `Errors`.
        :return:
        """
        if isinstance(self, Ok):
            return on_success(self.success)
        if isinstance(self, Errors):
            return on_error(self.errors)
        if isinstance(self, Panic):
            raise MultipleExceptions.merge(*self.exceptions, errors=self.errors)
        raise MatchError(f"{self} should be a Result")

    @final
    @safe
    def fold_raise(
        self, on_success: Callable[[A], X], on_error: Callable[[List[E]], X]
    ) -> X:
        """
        Transform this `Result[E,A]` into `X` if this result is an `Ok` or `Errors`.
        But raise the stored exceptions is this is a panic.

        It is useful to raise an exceptions on panics.

        :param on_success: is called if this result is a `Ok`.
        :param on_error: is called if this result is a `Errors`.
        :return:
        """
        return self.unsafe_fold_raise(on_success, on_error)

    @final
    def unsafe_flat_map(self, f: Callable[[A], Result[E, A2]]) -> Result[E, A2]:
        """
        The usual monadic operation called
            - bind, >>=: in Haskell
            - flatMap: in Scala
            - andThem: in Elm
            ...

        Chain operations returning results.

        :param f: operation to perform it this result is an `Ok`.
        :return: the result combined result.
        """
        if isinstance(self, Ok):
            return f(self.success)
        return self  # type: ignore

    @final
    @safe
    def flat_map(self, f: Callable[[A], Result[E, A2]]) -> Result[E, A2]:
        """
        The usual monadic operation called
            - bind, >>=: in Haskell
            - flatMap: in Scala
            - andThem: in Elm
            ...

        Chain operations returning results.

        :param f: operation to perform it this result is an `Ok`.
        :return: the result combined result.
        """
        return self.unsafe_flat_map(f)

    @final
    def unsafe_tri_map(
        self,
        f: Callable[[A], A2],
        g: Callable[[E], E2],
        h: Callable[[TracedException], TracedException],
    ) -> Result[E2, A2]:
        """
        Transform the value/errors/exceptions stored in this result.
        :param f: how to transform the value a if this result is `Ok(a)`
        :param g: how to transform the errors e if this result is `Errors(e)`
        :param h: how to transform the exceptions p if this result is `Panic(p)`
        :return: the "same" result with the stored value transformed.
        """
        return self.unsafe_fold(
            lambda x: Ok(f(x)),
            lambda x: Errors([g(y) for y in x]),
            lambda x, y: Panic(exceptions=[h(z) for z in x], errors=[g(z) for z in y]),
        )

    @final
    @safe
    def tri_map(
        self,
        f: Callable[[A], A2],
        g: Callable[[E], E2],
        h: Callable[[TracedException], TracedException],
    ) -> Result[E2, A2]:
        """
        Transform the value/errors/exceptions stored in this result.
        :param f: how to transform the value a if this result is `Ok(a)`
        :param g: how to transform the errors e if this result is `Errors(e)`
        :param h: how to transform the exceptions p if this result is `Panic(p)`
        :return: the "same" result with the stored value transformed.
        """
        return self.unsafe_tri_map(f, g, h)

    @final
    def is_ok(self) -> bool:
        """
        :return: True if this result is an `Ok`
        """
        return isinstance(self, Ok)

    @final
    def is_error(self) -> bool:
        """
        :return: True if this result is an `Errors`
        """
        return isinstance(self, Errors)

    @final
    def is_panic(self) -> bool:
        """
        :return: True if this result is an `Panic`
        """
        return isinstance(self, Panic)

    @final
    def unsafe_map(self, f: Callable[[A], A2]) -> Result[E, A2]:
        """
        Transform the value stored in `Ok`, it this result is an `Ok`.
        :param f: the transformation function.
        :return:
        """
        if isinstance(self, Ok):
            return Ok(f(self.success))
        return self  # type: ignore

    @final
    @safe
    def map(self, f: Callable[[A], A2]) -> Result[E, A2]:
        """
        Transform the value stored in `Ok`, it this result is an `Ok`.
        :param f: the transformation function.
        :return:
        """
        return self.unsafe_map(f)

    @final
    def zip(self: Result[E, A], *arg: Result[E, A]) -> Result[E, List[A]]:
        """
        Transform a list of Result (including self) into a Result of list.

        Is Ok is all results are Ok.
        Is Errors some are Ok, but at least one is an errors but no panics.
        Is Panic is there is at least one panic.
        """
        return zip((self, *arg))  # type: ignore

    @final
    def unsafe_ap(
        self: Result[E, Callable[[X], A]], *arg: Result[E, X]
    ) -> Result[E, A]:
        """
        Noting functions from X to A: `[X1, ..., Xn] -> A`.

        If this result represent a computation returning a function `f: [X1,...,XN] -> A`
        and arg represent a computation returning a value `x1: X1`,...,`xn: Xn`, then
        `self.ap(arg)` represents the computation returning `f(x1,...,xn): A`.
        """
        return zip((self, *arg)).unsafe_map(lambda l: l[0](*l[1:]))  # type: ignore

    @final
    @safe
    def ap(self: Result[E, Callable[[X], A]], *arg: Result[E, X]) -> Result[E, A]:
        """
        Noting functions from [X1,...,XN] to A: `[X1, ..., Xn] -> A`.

        If this result represent a computation returning a function `f: [X1,...,XN] -> A`
        and arg represent computations returning values `x1: X1`,...,`xn: Xn` then
        `self.ap(arg)` represents the computation returning `f(x1,...,xn): A`.
        """
        return self.unsafe_ap(*arg)  # type: ignore

    @final
    def flatten(self: Result[E, Result[E, A]]) -> Result[E, A]:
        """
        The concatenation function on results.
        """
        if isinstance(self, Ok):
            return self.success
        return self  # type: ignore

    @final
    def unsafe_map_error(self, f: Callable[[E], E2]) -> Result[E2, A]:
        """
        Transform the errors stored if this result is an `Errors`.
        :param f: the transformation function
        :return:
        """
        if isinstance(self, Errors):
            return Errors([f(e) for e in self.errors])
        return self  # type: ignore

    @final
    @safe
    def map_error(self, f: Callable[[E], E2]) -> Result[E2, A]:
        """
        Transform the errors stored if this result is an `Errors`.
        :param f: the transformation function
        :return:
        """
        return self.unsafe_map_error(f)

    @final
    def unsafe_catch(self, handler: Callable[[List[E]], Result[E, A]]) -> Result[E, A]:
        """
        React to errors (the except part of a try-except).

        If this result is an `Errors(some_error)`, then replace it with `handler(some_error)`.
        Otherwise, do nothing.
        """
        if isinstance(self, Errors):
            return handler(self.errors)
        return self

    @final
    @safe
    def catch(self, handler: Callable[[List[E]], Result[E, A]]) -> Result[E, A]:
        """
        React to errors (the except part of a try-except).

        If this result is an `Errors(some_error)`, then replace it with `handler(some_error)`.
        Otherwise, do nothing.
        """
        return self.unsafe_catch(handler)

    @final
    def unsafe_map_panic(
        self, f: Callable[[TracedException], TracedException]
    ) -> Result[E, A]:
        """
        Transform the exceptions stored if this result is a `Panic(some_exception)`.
        """
        if isinstance(self, Panic):
            return Panic(
                exceptions=[f(exn) for exn in self.exceptions], errors=self.errors
            )
        return self

    @final
    @safe
    def map_panic(
        self, f: Callable[[TracedException], TracedException]
    ) -> Result[E, A]:
        """
        Transform the exceptions stored if this result is a `Panic(some_exception)`.
        """
        return self.unsafe_map_panic(f)

    @final
    def unsafe_recover(
        self, handler: Callable[[List[TracedException], List[E]], Result[E, A]]
    ) -> Result[E, A]:
        """
        React to panics (the except part of a try-except).

        If this result is a `Panic(exceptions)`, replace it by `handler(exceptions)`.
        Otherwise do nothing.
        """
        if isinstance(self, Panic):
            return handler(self.exceptions, self.errors)
        return self

    @final
    @safe
    def recover(
        self, handler: Callable[[List[TracedException], List[E]], Result[E, A]]
    ) -> Result[E, A]:
        """
        React to panics (the except part of a try-except).

        If this result is a `Panic(exceptions)`, replace it by `handler(exceptions)`.
        Otherwise do nothing.
        """
        return self.unsafe_recover(handler)

    @final
    def raise_on_panic(self) -> Result[E, A]:
        """
        If this result is an `Ok` or `Errors`, do nothing.
        If it is a `Panic(some_exception)`, raise the exceptions.

        Use with extreme care since it raise exceptions.
        """
        if isinstance(self, Panic):
            raise MultipleExceptions.merge(*self.exceptions, errors=self.errors)
        return self

    @final
    def unsafe_get(self) -> Result[E, A]:
        """
        If this result is an `Ok`, do nothing.
        If it is a `Panic(some_exception)`, raise the exceptions.

        Use with extreme care since it raise exceptions.
        """
        if isinstance(self, Ok):
            return self.success
        if isinstance(self, Errors):
            raise DomainErrors(self.errors)
        if isinstance(self, Panic):
            raise MultipleExceptions.merge(*self.exceptions, errors=self.errors)
        raise MatchError(f"{self} should be a result.")

    @final
    def to_computation_status(self) -> ComputationStatus:
        """
        Transform this Result into a Computation Status
        :return:
        """
        if self.is_ok():
            return ComputationStatus.SUCCEEDED
        return ComputationStatus.FAILED

Ancestors

• typing.Generic

Subclasses

• Errors
• Ok
• Panic

Methods

def ap(*args: Any, **kwargs: Any) ‑> Any

Expand source code

def wrapper(*args: Any, **kwargs: Any) -> Any:
    try:
        return f(*args, **kwargs)
    except Exception as exception:
        return Panic(
            exceptions=[TracedException.in_except_clause(exception)], errors=[]
        )

def catch(*args: Any, **kwargs: Any) ‑> Any

Expand source code

def wrapper(*args: Any, **kwargs: Any) -> Any:
    try:
        return f(*args, **kwargs)
    except Exception as exception:
        return Panic(
            exceptions=[TracedException.in_except_clause(exception)], errors=[]
        )

def flat_map(*args: Any, **kwargs: Any) ‑> Any

Expand source code

def wrapper(*args: Any, **kwargs: Any) -> Any:
    try:
        return f(*args, **kwargs)
    except Exception as exception:
        return Panic(
            exceptions=[TracedException.in_except_clause(exception)], errors=[]
        )

def flatten(self: Result[E, Result[E, A]]) ‑> Result[~E, ~A]

The concatenation function on results.

Expand source code

@final
def flatten(self: Result[E, Result[E, A]]) -> Result[E, A]:
    """
    The concatenation function on results.
    """
    if isinstance(self, Ok):
        return self.success
    return self  # type: ignore

def fold(*args: Any, **kwargs: Any) ‑> Any

Expand source code

def wrapper(*args: Any, **kwargs: Any) -> Any:
    try:
        return f(*args, **kwargs)
    except Exception as exception:
        return Panic(
            exceptions=[TracedException.in_except_clause(exception)], errors=[]
        )

def fold_raise(*args: Any, **kwargs: Any) ‑> Any

Expand source code

def wrapper(*args: Any, **kwargs: Any) -> Any:
    try:
        return f(*args, **kwargs)
    except Exception as exception:
        return Panic(
            exceptions=[TracedException.in_except_clause(exception)], errors=[]
        )

def is_error(self) ‑> bool

:return: True if this result is an Errors

Expand source code

@final
def is_error(self) -> bool:
    """
    :return: True if this result is an `Errors`
    """
    return isinstance(self, Errors)

def is_ok(self) ‑> bool

:return: True if this result is an Ok

Expand source code

@final
def is_ok(self) -> bool:
    """
    :return: True if this result is an `Ok`
    """
    return isinstance(self, Ok)

def is_panic(self) ‑> bool

:return: True if this result is an Panic

Expand source code

@final
def is_panic(self) -> bool:
    """
    :return: True if this result is an `Panic`
    """
    return isinstance(self, Panic)

def map(*args: Any, **kwargs: Any) ‑> Any

Expand source code

def wrapper(*args: Any, **kwargs: Any) -> Any:
    try:
        return f(*args, **kwargs)
    except Exception as exception:
        return Panic(
            exceptions=[TracedException.in_except_clause(exception)], errors=[]
        )

def map_error(*args: Any, **kwargs: Any) ‑> Any

Expand source code

def wrapper(*args: Any, **kwargs: Any) -> Any:
    try:
        return f(*args, **kwargs)
    except Exception as exception:
        return Panic(
            exceptions=[TracedException.in_except_clause(exception)], errors=[]
        )

def map_panic(*args: Any, **kwargs: Any) ‑> Any

Expand source code

def wrapper(*args: Any, **kwargs: Any) -> Any:
    try:
        return f(*args, **kwargs)
    except Exception as exception:
        return Panic(
            exceptions=[TracedException.in_except_clause(exception)], errors=[]
        )

def raise_on_panic(self) ‑> Result[~E, ~A]

If this result is an Ok or Errors, do nothing. If it is a Panic(some_exception), raise the exceptions.

Use with extreme care since it raise exceptions.

Expand source code

@final
def raise_on_panic(self) -> Result[E, A]:
    """
    If this result is an `Ok` or `Errors`, do nothing.
    If it is a `Panic(some_exception)`, raise the exceptions.

    Use with extreme care since it raise exceptions.
    """
    if isinstance(self, Panic):
        raise MultipleExceptions.merge(*self.exceptions, errors=self.errors)
    return self

def recover(*args: Any, **kwargs: Any) ‑> Any

Expand source code

def wrapper(*args: Any, **kwargs: Any) -> Any:
    try:
        return f(*args, **kwargs)
    except Exception as exception:
        return Panic(
            exceptions=[TracedException.in_except_clause(exception)], errors=[]
        )

def to_computation_status(self) ‑> ComputationStatus

Transform this Result into a Computation Status :return:

Expand source code

@final
def to_computation_status(self) -> ComputationStatus:
    """
    Transform this Result into a Computation Status
    :return:
    """
    if self.is_ok():
        return ComputationStatus.SUCCEEDED
    return ComputationStatus.FAILED

def tri_map(*args: Any, **kwargs: Any) ‑> Any

Expand source code

def wrapper(*args: Any, **kwargs: Any) -> Any:
    try:
        return f(*args, **kwargs)
    except Exception as exception:
        return Panic(
            exceptions=[TracedException.in_except_clause(exception)], errors=[]
        )

def unsafe_ap(self: Result[E, Callable[[X], A]], *arg: Result[E, X]) ‑> Result[~E, ~A]

Noting functions from X to A: [X1, ..., Xn] -> A.

If this result represent a computation returning a function f: [X1,...,XN] -> A and arg represent a computation returning a value x1: X1,…,xn: Xn, then self.ap(arg) represents the computation returning f(x1,...,xn): A.

Expand source code

@final
def unsafe_ap(
    self: Result[E, Callable[[X], A]], *arg: Result[E, X]
) -> Result[E, A]:
    """
    Noting functions from X to A: `[X1, ..., Xn] -> A`.

    If this result represent a computation returning a function `f: [X1,...,XN] -> A`
    and arg represent a computation returning a value `x1: X1`,...,`xn: Xn`, then
    `self.ap(arg)` represents the computation returning `f(x1,...,xn): A`.
    """
    return zip((self, *arg)).unsafe_map(lambda l: l[0](*l[1:]))  # type: ignore

def unsafe_catch(self, handler: Callable[[List[E]], Result[E, A]]) ‑> Result[~E, ~A]

React to errors (the except part of a try-except).

If this result is an Errors(some_error), then replace it with handler(some_error). Otherwise, do nothing.

Expand source code

@final
def unsafe_catch(self, handler: Callable[[List[E]], Result[E, A]]) -> Result[E, A]:
    """
    React to errors (the except part of a try-except).

    If this result is an `Errors(some_error)`, then replace it with `handler(some_error)`.
    Otherwise, do nothing.
    """
    if isinstance(self, Errors):
        return handler(self.errors)
    return self

def unsafe_flat_map(self, f: Callable[[A], Result[E, A2]]) ‑> Result[~E, ~A2]

The usual monadic operation called - bind, >>=: in Haskell - flatMap: in Scala - andThem: in Elm …

Chain operations returning results.

:param f: operation to perform it this result is an Ok. :return: the result combined result.

Expand source code

@final
def unsafe_flat_map(self, f: Callable[[A], Result[E, A2]]) -> Result[E, A2]:
    """
    The usual monadic operation called
        - bind, >>=: in Haskell
        - flatMap: in Scala
        - andThem: in Elm
        ...

    Chain operations returning results.

    :param f: operation to perform it this result is an `Ok`.
    :return: the result combined result.
    """
    if isinstance(self, Ok):
        return f(self.success)
    return self  # type: ignore

def unsafe_fold(self, on_success: Callable[[A], X], on_error: Callable[[List[E]], X], on_panic: Callable[[List[TracedException], List[E]], X]) ‑> ~X

Transform this Result[E,A] into X. :param on_success: is called if this result is a Ok. :param on_error: is called if this result is a Errors. :param on_panic: is called if this result is a Panic. :return:

Expand source code

@final
def unsafe_fold(
    self,
    on_success: Callable[[A], X],
    on_error: Callable[[List[E]], X],
    on_panic: Callable[[List[TracedException], List[E]], X],
) -> X:
    """
    Transform this Result[E,A] into X.
    :param on_success: is called if this result is a `Ok`.
    :param on_error: is called if this result is a `Errors`.
    :param on_panic: is called if this result is a `Panic`.
    :return:
    """
    if isinstance(self, Ok):
        return on_success(self.success)
    if isinstance(self, Errors):
        return on_error(self.errors)
    if isinstance(self, Panic):
        return on_panic(self.exceptions, self.errors)
    raise on_panic([MatchError(f"{self} should be a Result")], [])

def unsafe_fold_raise(self, on_success: Callable[[A], X], on_error: Callable[[List[E]], X]) ‑> ~X

Transform this Result[E,A] into X if this result is an Ok or Errors. But raise the stored exceptions is this is a panic.

It is useful to raise an exceptions on panics.

:param on_success: is called if this result is a Ok. :param on_error: is called if this result is a Errors. :return:

Expand source code

@final
def unsafe_fold_raise(
    self, on_success: Callable[[A], X], on_error: Callable[[List[E]], X]
) -> X:
    """
    Transform this `Result[E,A]` into `X` if this result is an `Ok` or `Errors`.
    But raise the stored exceptions is this is a panic.

    It is useful to raise an exceptions on panics.

    :param on_success: is called if this result is a `Ok`.
    :param on_error: is called if this result is a `Errors`.
    :return:
    """
    if isinstance(self, Ok):
        return on_success(self.success)
    if isinstance(self, Errors):
        return on_error(self.errors)
    if isinstance(self, Panic):
        raise MultipleExceptions.merge(*self.exceptions, errors=self.errors)
    raise MatchError(f"{self} should be a Result")

def unsafe_get(self) ‑> Result[~E, ~A]

If this result is an Ok, do nothing. If it is a Panic(some_exception), raise the exceptions.

Use with extreme care since it raise exceptions.

Expand source code

@final
def unsafe_get(self) -> Result[E, A]:
    """
    If this result is an `Ok`, do nothing.
    If it is a `Panic(some_exception)`, raise the exceptions.

    Use with extreme care since it raise exceptions.
    """
    if isinstance(self, Ok):
        return self.success
    if isinstance(self, Errors):
        raise DomainErrors(self.errors)
    if isinstance(self, Panic):
        raise MultipleExceptions.merge(*self.exceptions, errors=self.errors)
    raise MatchError(f"{self} should be a result.")

def unsafe_map(self, f: Callable[[A], A2]) ‑> Result[~E, ~A2]

Transform the value stored in Ok, it this result is an Ok. :param f: the transformation function. :return:

Expand source code

@final
def unsafe_map(self, f: Callable[[A], A2]) -> Result[E, A2]:
    """
    Transform the value stored in `Ok`, it this result is an `Ok`.
    :param f: the transformation function.
    :return:
    """
    if isinstance(self, Ok):
        return Ok(f(self.success))
    return self  # type: ignore

def unsafe_map_error(self, f: Callable[[E], E2]) ‑> Result[~E2, ~A]

Transform the errors stored if this result is an Errors. :param f: the transformation function :return:

Expand source code

@final
def unsafe_map_error(self, f: Callable[[E], E2]) -> Result[E2, A]:
    """
    Transform the errors stored if this result is an `Errors`.
    :param f: the transformation function
    :return:
    """
    if isinstance(self, Errors):
        return Errors([f(e) for e in self.errors])
    return self  # type: ignore

def unsafe_map_panic(self, f: Callable[[TracedException], TracedException]) ‑> Result[~E, ~A]

Transform the exceptions stored if this result is a Panic(some_exception).

Expand source code

@final
def unsafe_map_panic(
    self, f: Callable[[TracedException], TracedException]
) -> Result[E, A]:
    """
    Transform the exceptions stored if this result is a `Panic(some_exception)`.
    """
    if isinstance(self, Panic):
        return Panic(
            exceptions=[f(exn) for exn in self.exceptions], errors=self.errors
        )
    return self

def unsafe_recover(self, handler: Callable[[List[TracedException], List[E]], Result[E, A]]) ‑> Result[~E, ~A]

React to panics (the except part of a try-except).

If this result is a Panic(exceptions), replace it by handler(exceptions). Otherwise do nothing.

Expand source code

@final
def unsafe_recover(
    self, handler: Callable[[List[TracedException], List[E]], Result[E, A]]
) -> Result[E, A]:
    """
    React to panics (the except part of a try-except).

    If this result is a `Panic(exceptions)`, replace it by `handler(exceptions)`.
    Otherwise do nothing.
    """
    if isinstance(self, Panic):
        return handler(self.exceptions, self.errors)
    return self

def unsafe_tri_map(self, f: Callable[[A], A2], g: Callable[[E], E2], h: Callable[[TracedException], TracedException]) ‑> Result[~E2, ~A2]

Transform the value/errors/exceptions stored in this result. :param f: how to transform the value a if this result is Ok(a) :param g: how to transform the errors e if this result is Errors(e) :param h: how to transform the exceptions p if this result is Panic(p) :return: the "same" result with the stored value transformed.

Expand source code

@final
def unsafe_tri_map(
    self,
    f: Callable[[A], A2],
    g: Callable[[E], E2],
    h: Callable[[TracedException], TracedException],
) -> Result[E2, A2]:
    """
    Transform the value/errors/exceptions stored in this result.
    :param f: how to transform the value a if this result is `Ok(a)`
    :param g: how to transform the errors e if this result is `Errors(e)`
    :param h: how to transform the exceptions p if this result is `Panic(p)`
    :return: the "same" result with the stored value transformed.
    """
    return self.unsafe_fold(
        lambda x: Ok(f(x)),
        lambda x: Errors([g(y) for y in x]),
        lambda x, y: Panic(exceptions=[h(z) for z in x], errors=[g(z) for z in y]),
    )

def zip(self: Result[E, A], *arg: Result[E, A]) ‑> Result[~E, typing.List[~A]]

Transform a list of Result (including self) into a Result of list.

Is Ok is all results are Ok. Is Errors some are Ok, but at least one is an errors but no panics. Is Panic is there is at least one panic.

Expand source code

@final
def zip(self: Result[E, A], *arg: Result[E, A]) -> Result[E, List[A]]:
    """
    Transform a list of Result (including self) into a Result of list.

    Is Ok is all results are Ok.
    Is Errors some are Ok, but at least one is an errors but no panics.
    Is Panic is there is at least one panic.
    """
    return zip((self, *arg))  # type: ignore

@final
@dataclass
class TracedException:
    __slots__ = ("exception", "stack_trace")

    exception: Exception
    """
    The exception that was raised.
    """

    stack_trace: str
    """
    Its stack trace.
    """

    def __str__(self):
        return f"{self.exception}\n{self.stack_trace}"

    @classmethod
    def in_except_clause(cls, exn: Exception) -> TracedException:
        """
        Collect the stack trace of the exception.

        BEWARE: this method should only be used in the except clause
        of a try-except block and called with the caught exception!

        :param exn:
        :return:
        """
        if isinstance(exn, TracedException):
            return exn
        return TracedException(exception=exn, stack_trace=format_exc())

    @classmethod
    def with_stack_trace(cls, exn: Exception) -> TracedException:
        """
        Collect the stack trace at the current position.

        :param exn:
        :return:
        """
        if isinstance(exn, TracedException):
            return exn
        return TracedException(exception=exn, stack_trace="".join(format_stack()))

    @classmethod
    def ensure_traced(cls, exception: Exception) -> TracedException:
        return cls.with_stack_trace(exception)

    @classmethod
    def ensure_list_traced(
        cls, exceptions: Iterable[Exception]
    ) -> List[TracedException]:
        return [cls.ensure_traced(exn) for exn in exceptions]

Static methods

def ensure_list_traced(exceptions: Iterable[Exception]) ‑> List[TracedException]

Expand source code

@classmethod
def ensure_list_traced(
    cls, exceptions: Iterable[Exception]
) -> List[TracedException]:
    return [cls.ensure_traced(exn) for exn in exceptions]

def ensure_traced(exception: Exception) ‑> TracedException

Expand source code

@classmethod
def ensure_traced(cls, exception: Exception) -> TracedException:
    return cls.with_stack_trace(exception)

def in_except_clause(exn: Exception) ‑> TracedException

Collect the stack trace of the exception.

BEWARE: this method should only be used in the except clause of a try-except block and called with the caught exception!

:param exn: :return:

Expand source code

@classmethod
def in_except_clause(cls, exn: Exception) -> TracedException:
    """
    Collect the stack trace of the exception.

    BEWARE: this method should only be used in the except clause
    of a try-except block and called with the caught exception!

    :param exn:
    :return:
    """
    if isinstance(exn, TracedException):
        return exn
    return TracedException(exception=exn, stack_trace=format_exc())

def with_stack_trace(exn: Exception) ‑> TracedException

Collect the stack trace at the current position.

:param exn: :return:

Expand source code

@classmethod
def with_stack_trace(cls, exn: Exception) -> TracedException:
    """
    Collect the stack trace at the current position.

    :param exn:
    :return:
    """
    if isinstance(exn, TracedException):
        return exn
    return TracedException(exception=exn, stack_trace="".join(format_stack()))

Instance variables

var exception : Exception

The exception that was raised.

var stack_trace : str

Its stack trace.

@final
@dataclass
class UpdateResult(Generic[A, B]):
    """
    The result of the `update` methods of `Var`.
    """

    __slots__ = "old_value", "new_value", "returned"

    """
    The result of an Update
    """

    old_value: A
    """
    The value before the Update
    """

    new_value: A
    """
    The value after the update
    """

    returned: B
    """
    The result produced by the update
    """

Ancestors

• typing.Generic

Instance variables

var new_value : ~A

The value after the update

var old_value : ~A

The value before the Update

var returned : ~B

The result produced by the update

@final
@dataclass
class Val(Generic[A]):
    """
    Immutable Value.

    Used to create local "variables" in lambdas.
    """

    __slots__ = "value"

    value: A

    def get(self) -> A:
        """
        Get this Val value.
        :return:
        """
        return self.value

    def get_io(self) -> IO[None, None, A]:
        """
        Get this Val value.
        :return:
        """
        return io.defer(self.get)

    def get_rs(self) -> Resource[None, None, A]:
        """
        Get this Val value.
        :return:
        """
        return resource.defer(self.get)

    @classmethod
    def pure(cls, a: A) -> Val[A]:
        """
        Create a new Val with value `a`

        :param a: the value of this val.
        :return:
        """
        return Val(a)

    def map(self, f: Callable[[A], B]) -> Val[B]:
        """
        Create a new Val from this one by applying this **pure** function.

        :param f:
        :return:
        """
        return Val(f(self.value))

    def traverse(self, f: Callable[[A], IO[Any, Any, B]]) -> IO[Any, Any, Val[B]]:
        """
        Create a new Val from this one by applying this `IO` function.

        :param f:
        :return:
        """

        return io.defer_io(f, self.value).map(Val)

    def flat_map(self, f: Callable[[A], Val[B]]) -> Val[B]:
        """
        Create a new Val from this one.

        :param f:
        :return:
        """
        return f(self.value)

    def flatten(self: Val[Val[B]]) -> Val[B]:  # A = Val[B]
        """ "
        Flatten this `Val[Val[A]]` into a `Val[A]`
        """

        return Val(self.value.value)

    @classmethod
    def zip(cls, *vals: Val[A]) -> Val[List[A]]:
        """ "
        Group these list of Val into a Val of List
        """

        if len(vals) == 1 and isinstance(vals[0], abc.Iterable):
            return Val([x.value for x in vals[0]])
        return Val([x.value for x in vals])  # type: ignore

    def zip_with(self, *vals: Any) -> Val[List[A]]:
        """
        Group this Val with other Val into a list of Val.

        :param vals: other Val to combine with self.
        :return:
        """

        return Val.zip(self, *vals)

    def ap(self, *arg: Val[A]) -> Val[B]:
        """
        Apply the function contained in this Val to `args` Vals.

        :param arg:
        :return:
        """

        if len(arg) == 1 and isinstance(arg[0], abc.Iterable):
            l = [x.value for x in arg[0]]
        else:
            l = [x.value for x in arg]
        return Val(self.value(*l))  # type: ignore

Ancestors

• typing.Generic

Static methods

def pure(a: A) ‑> Val[~A]

Create a new Val with value a

:param a: the value of this val. :return:

Expand source code

@classmethod
def pure(cls, a: A) -> Val[A]:
    """
    Create a new Val with value `a`

    :param a: the value of this val.
    :return:
    """
    return Val(a)

def zip(*vals: Val[A]) ‑> Val[typing.List[~A]]

" Group these list of Val into a Val of List

Expand source code

@classmethod
def zip(cls, *vals: Val[A]) -> Val[List[A]]:
    """ "
    Group these list of Val into a Val of List
    """

    if len(vals) == 1 and isinstance(vals[0], abc.Iterable):
        return Val([x.value for x in vals[0]])
    return Val([x.value for x in vals])  # type: ignore

Instance variables

var value : ~A

Return an attribute of instance, which is of type owner.

Methods

def ap(self, *arg: Val[A]) ‑> Val[~B]

Apply the function contained in this Val to args Vals.

:param arg: :return:

Expand source code

def ap(self, *arg: Val[A]) -> Val[B]:
    """
    Apply the function contained in this Val to `args` Vals.

    :param arg:
    :return:
    """

    if len(arg) == 1 and isinstance(arg[0], abc.Iterable):
        l = [x.value for x in arg[0]]
    else:
        l = [x.value for x in arg]
    return Val(self.value(*l))  # type: ignore

def flat_map(self, f: Callable[[A], Val[B]]) ‑> Val[~B]

Create a new Val from this one.

:param f: :return:

Expand source code

def flat_map(self, f: Callable[[A], Val[B]]) -> Val[B]:
    """
    Create a new Val from this one.

    :param f:
    :return:
    """
    return f(self.value)

def flatten(self: Val[Val[B]]) ‑> Val[~B]

" Flatten this Val[Val[A]] into a Val[A]

Expand source code

def flatten(self: Val[Val[B]]) -> Val[B]:  # A = Val[B]
    """ "
    Flatten this `Val[Val[A]]` into a `Val[A]`
    """

    return Val(self.value.value)

def get(self) ‑> ~A

Get this Val value. :return:

Expand source code

def get(self) -> A:
    """
    Get this Val value.
    :return:
    """
    return self.value

def get_io(self) ‑> IO[NoneType, NoneType, ~A]

Get this Val value. :return:

Expand source code

def get_io(self) -> IO[None, None, A]:
    """
    Get this Val value.
    :return:
    """
    return io.defer(self.get)

def get_rs(self) ‑> Resource[NoneType, NoneType, ~A]

Get this Val value. :return:

Expand source code

def get_rs(self) -> Resource[None, None, A]:
    """
    Get this Val value.
    :return:
    """
    return resource.defer(self.get)

def map(self, f: Callable[[A], B]) ‑> Val[~B]

Create a new Val from this one by applying this pure function.

:param f: :return:

Expand source code

def map(self, f: Callable[[A], B]) -> Val[B]:
    """
    Create a new Val from this one by applying this **pure** function.

    :param f:
    :return:
    """
    return Val(f(self.value))

def traverse(self, f: Callable[[A], IO[Any, Any, B]]) ‑> IO[typing.Any, typing.Any, Val[~B]]

Create a new Val from this one by applying this IO function.

:param f: :return:

Expand source code

def traverse(self, f: Callable[[A], IO[Any, Any, B]]) -> IO[Any, Any, Val[B]]:
    """
    Create a new Val from this one by applying this `IO` function.

    :param f:
    :return:
    """

    return io.defer_io(f, self.value).map(Val)

def zip_with(self, *vals: Any) ‑> Val[typing.List[~A]]

Group this Val with other Val into a list of Val.

:param vals: other Val to combine with self. :return:

Expand source code

def zip_with(self, *vals: Any) -> Val[List[A]]:
    """
    Group this Val with other Val into a list of Val.

    :param vals: other Val to combine with self.
    :return:
    """

    return Val.zip(self, *vals)

@final
class Var(Generic[A]):
    """
    A mutable variable.
    **All** concurrent access are protected using a Reentrant Lock.

    **IMPORTANT:** /!\\ **NEVER CREATE VARIABLES BY THE CONSTRUCTOT !!!!** /!\\

    Use `Var.create instead.`
    """

    __slots__ = "_lock", "_value"

    def __init__(self, lock: Resource[Any, None, None], value: A):
        self._lock = lock
        self._value = value

    def lock(self) -> Resource[Any, None, None]:
        """
        The Reentrant Lock that guarantees exclusive access to this variable.
        """
        return self._lock

    @classmethod
    def create(cls, a: A) -> IO[Any, None, Var[A]]:
        """
        Create a new variable whose value is `a`.

        :param a:
        :return:
        """
        return resource.reentrant_lock.map(lambda lock: Var(lock, a))

    @classmethod
    def create_rs(cls, a: A) -> Resource[Any, None, Var[A]]:
        """
        Create a new variable whose value is `a`.

        :param a:
        :return:
        """
        return resource.lift_io(resource.reentrant_lock.map(lambda lock: Var(lock, a)))

    #############
    #   GETTER  #
    #############

    def get(self) -> IO[None, None, A]:
        """
        Get the current value of this variable.
        :return:
        """
        return self._lock.with_(io.defer(lambda: self._value))

    def get_rs(self) -> Resource[None, None, A]:
        """
        Get the current value of this variable.
        :return:
        """
        return self._lock.then(resource.defer(lambda: self._value))

    #############
    #   SETTER  #
    #############

    def set(self, v: A) -> IO[None, None, None]:
        """
        Assign a new value to this variable.

        :param v:
        :return:
        """

        def h(w) -> None:
            self._value = w

        return self._lock.with_(io.defer(h, v))

    def set_rs(self, v: A) -> Resource[None, None, None]:
        """
        Assign a new value to this variable.

        :param v:
        :return:
        """

        def h(w) -> None:
            self._value = w

        return self._lock.then(resource.defer(h, v))

    #####################
    #   GETTER + SETTER #
    #####################

    def get_and_set(self, v: A) -> IO[Any, None, A]:
        """
        Assign a new value to this variable. The previous value is returned.

        :param v:
        :return:
        """

        def h():
            old_value = self._value
            self._value = v
            return old_value

        return self._lock.with_(io.defer(h))

    def get_and_set_rs(self, v: A) -> Resource[None, None, None]:
        """
        Assign a new value to this variable. The previous value is returned.

        :param v:
        :return:
        """

        def h():
            old_value = self._value
            self._value = v
            return old_value

        return self._lock.then(resource.defer(h))

    ###############
    #    UPDATE   #
    ###############

    def update(self, f: Callable[[A], Tuple[A, B]]) -> IO[Any, Any, UpdateResult[A, B]]:
        """
        Update the value contained in this variable.

        :param f:
        :return:
        """

        def h() -> UpdateResult[A, B]:
            old_value = self._value
            new_value, ret = f(self._value)
            self._value = new_value
            return UpdateResult(old_value, new_value, ret)

        return self._lock.with_(io.defer(h))

    def update_io(
        self, f: Callable[[A], IO[Any, None, Tuple[A, B]]]
    ) -> IO[Any, None, UpdateResult[A, B]]:
        """
        Update the value contained in this variable.

        :param f:
        :return:
        """

        def h() -> IO[Any, None, UpdateResult[A, B]]:
            old_value = self._value

            def g(x: Tuple[A, B]) -> UpdateResult[A, B]:
                self._value = x[0]
                return UpdateResult(old_value, self._value, x[1])

            return f(old_value).map(g)

        return self._lock.with_(io.defer_io(h))

    def update_rs(
        self, f: Callable[[A], Resource[Any, None, Tuple[A, B]]]
    ) -> Resource[Any, None, UpdateResult[A, B]]:
        """
        Update the value contained in this variable.

        :param f:
        :return:
        """

        def h() -> Resource[Any, None, UpdateResult[A, B]]:
            old_value = self._value

            def g(x: Tuple[A, B]) -> UpdateResult[A, B]:
                self._value = x[0]
                return UpdateResult(old_value, self._value, x[1])

            return f(old_value).map(g)

        return self._lock.then(resource.defer_resource(h))

    ######################
    #  Creating New Vars #
    ######################

    def traverse(self, f: Callable[[A], IO[Any, None, B]]) -> IO[Any, None, Var[B]]:
        """
        Create a new variable by transforming the current value of this variable.

        :param f:
        :return:
        """

        def h():
            return f(self._value).flat_map(Var.create)

        return self._lock.with_(io.defer_io(h))

    @classmethod
    def zip(cls, *vars: Var[A]) -> IO[Any, None, List[A]]:
        """ "
        Group these variables current values into a list.
        """

        if len(vars) == 1 and isinstance(vars[0], abc.Iterable):
            args = vars[0]
        else:
            args = vars
        return resource.zip([x._lock for x in args]).with_(
            io.defer(lambda: [x._value for x in args])
        )

    def zip_with(self, *vars: Var[A]) -> IO[Any, None, List[A]]:
        """
        Group this variable current value with `vars` variable current values.

        :param vals: other variables to combine with self.
        :return:
        """
        return Var.zip(self, *vars)

    def ap(self, *arg: Var[A]) -> IO[Any, None, B]:
        """
        Apply the function contained in this variable to `args` variables.

        :param arg:
        :return:
        """
        return self.zip_with(*arg).map(lambda l: l[0](*l[1:]))  # type: ignore

Ancestors

• typing.Generic

Static methods

def create(a: A) ‑> IO[typing.Any, NoneType, Var[~A]]

Create a new variable whose value is a.

:param a: :return:

Expand source code

@classmethod
def create(cls, a: A) -> IO[Any, None, Var[A]]:
    """
    Create a new variable whose value is `a`.

    :param a:
    :return:
    """
    return resource.reentrant_lock.map(lambda lock: Var(lock, a))

def create_rs(a: A) ‑> Resource[typing.Any, NoneType, Var[~A]]

Create a new variable whose value is a.

:param a: :return:

Expand source code

@classmethod
def create_rs(cls, a: A) -> Resource[Any, None, Var[A]]:
    """
    Create a new variable whose value is `a`.

    :param a:
    :return:
    """
    return resource.lift_io(resource.reentrant_lock.map(lambda lock: Var(lock, a)))

def zip(*vars: Var[A]) ‑> IO[typing.Any, NoneType, typing.List[~A]]

" Group these variables current values into a list.

Expand source code

@classmethod
def zip(cls, *vars: Var[A]) -> IO[Any, None, List[A]]:
    """ "
    Group these variables current values into a list.
    """

    if len(vars) == 1 and isinstance(vars[0], abc.Iterable):
        args = vars[0]
    else:
        args = vars
    return resource.zip([x._lock for x in args]).with_(
        io.defer(lambda: [x._value for x in args])
    )

Methods

def ap(self, *arg: Var[A]) ‑> IO[typing.Any, NoneType, ~B]

Apply the function contained in this variable to args variables.

:param arg: :return:

Expand source code

def ap(self, *arg: Var[A]) -> IO[Any, None, B]:
    """
    Apply the function contained in this variable to `args` variables.

    :param arg:
    :return:
    """
    return self.zip_with(*arg).map(lambda l: l[0](*l[1:]))  # type: ignore

def get(self) ‑> IO[NoneType, NoneType, ~A]

Get the current value of this variable. :return:

Expand source code

def get(self) -> IO[None, None, A]:
    """
    Get the current value of this variable.
    :return:
    """
    return self._lock.with_(io.defer(lambda: self._value))

def get_and_set(self, v: A) ‑> IO[typing.Any, NoneType, ~A]

Assign a new value to this variable. The previous value is returned.

:param v: :return:

Expand source code

def get_and_set(self, v: A) -> IO[Any, None, A]:
    """
    Assign a new value to this variable. The previous value is returned.

    :param v:
    :return:
    """

    def h():
        old_value = self._value
        self._value = v
        return old_value

    return self._lock.with_(io.defer(h))

def get_and_set_rs(self, v: A) ‑> Resource[NoneType, NoneType, NoneType]

Assign a new value to this variable. The previous value is returned.

:param v: :return:

Expand source code

def get_and_set_rs(self, v: A) -> Resource[None, None, None]:
    """
    Assign a new value to this variable. The previous value is returned.

    :param v:
    :return:
    """

    def h():
        old_value = self._value
        self._value = v
        return old_value

    return self._lock.then(resource.defer(h))

def get_rs(self) ‑> Resource[NoneType, NoneType, ~A]

Get the current value of this variable. :return:

Expand source code

def get_rs(self) -> Resource[None, None, A]:
    """
    Get the current value of this variable.
    :return:
    """
    return self._lock.then(resource.defer(lambda: self._value))

def lock(self) ‑> Resource[typing.Any, NoneType, NoneType]

The Reentrant Lock that guarantees exclusive access to this variable.

Expand source code

def lock(self) -> Resource[Any, None, None]:
    """
    The Reentrant Lock that guarantees exclusive access to this variable.
    """
    return self._lock

def set(self, v: A) ‑> IO[NoneType, NoneType, NoneType]

Assign a new value to this variable.

:param v: :return:

Expand source code

def set(self, v: A) -> IO[None, None, None]:
    """
    Assign a new value to this variable.

    :param v:
    :return:
    """

    def h(w) -> None:
        self._value = w

    return self._lock.with_(io.defer(h, v))

def set_rs(self, v: A) ‑> Resource[NoneType, NoneType, NoneType]

Assign a new value to this variable.

:param v: :return:

Expand source code

def set_rs(self, v: A) -> Resource[None, None, None]:
    """
    Assign a new value to this variable.

    :param v:
    :return:
    """

    def h(w) -> None:
        self._value = w

    return self._lock.then(resource.defer(h, v))

def traverse(self, f: Callable[[A], IO[Any, None, B]]) ‑> IO[typing.Any, NoneType, Var[~B]]

Create a new variable by transforming the current value of this variable.

:param f: :return:

Expand source code

def traverse(self, f: Callable[[A], IO[Any, None, B]]) -> IO[Any, None, Var[B]]:
    """
    Create a new variable by transforming the current value of this variable.

    :param f:
    :return:
    """

    def h():
        return f(self._value).flat_map(Var.create)

    return self._lock.with_(io.defer_io(h))

def update(self, f: Callable[[A], Tuple[A, B]]) ‑> IO[typing.Any, typing.Any, UpdateResult[~A, ~B]]

Update the value contained in this variable.

:param f: :return:

Expand source code

def update(self, f: Callable[[A], Tuple[A, B]]) -> IO[Any, Any, UpdateResult[A, B]]:
    """
    Update the value contained in this variable.

    :param f:
    :return:
    """

    def h() -> UpdateResult[A, B]:
        old_value = self._value
        new_value, ret = f(self._value)
        self._value = new_value
        return UpdateResult(old_value, new_value, ret)

    return self._lock.with_(io.defer(h))

def update_io(self, f: Callable[[A], IO[Any, None, Tuple[A, B]]]) ‑> IO[typing.Any, NoneType, UpdateResult[~A, ~B]]

Update the value contained in this variable.

:param f: :return:

Expand source code

def update_io(
    self, f: Callable[[A], IO[Any, None, Tuple[A, B]]]
) -> IO[Any, None, UpdateResult[A, B]]:
    """
    Update the value contained in this variable.

    :param f:
    :return:
    """

    def h() -> IO[Any, None, UpdateResult[A, B]]:
        old_value = self._value

        def g(x: Tuple[A, B]) -> UpdateResult[A, B]:
            self._value = x[0]
            return UpdateResult(old_value, self._value, x[1])

        return f(old_value).map(g)

    return self._lock.with_(io.defer_io(h))

def update_rs(self, f: Callable[[A], Resource[Any, None, Tuple[A, B]]]) ‑> Resource[typing.Any, NoneType, UpdateResult[~A, ~B]]

Update the value contained in this variable.

:param f: :return:

Expand source code

def update_rs(
    self, f: Callable[[A], Resource[Any, None, Tuple[A, B]]]
) -> Resource[Any, None, UpdateResult[A, B]]:
    """
    Update the value contained in this variable.

    :param f:
    :return:
    """

    def h() -> Resource[Any, None, UpdateResult[A, B]]:
        old_value = self._value

        def g(x: Tuple[A, B]) -> UpdateResult[A, B]:
            self._value = x[0]
            return UpdateResult(old_value, self._value, x[1])

        return f(old_value).map(g)

    return self._lock.then(resource.defer_resource(h))

def zip_with(self, *vars: Var[A]) ‑> IO[typing.Any, NoneType, typing.List[~A]]

Group this variable current value with vars variable current values.

:param vals: other variables to combine with self. :return:

Expand source code

def zip_with(self, *vars: Var[A]) -> IO[Any, None, List[A]]:
    """
    Group this variable current value with `vars` variable current values.

    :param vals: other variables to combine with self.
    :return:
    """
    return Var.zip(self, *vars)