Source code for returns.future

from functools import wraps
from typing import (
    Any,
    Awaitable,
    Callable,
    Coroutine,
    Generator,
    Iterable,
    Sequence,
    Type,
    TypeVar,
)

from typing_extensions import final

from returns._internal.futures import _future, _future_result
from returns._internal.iterable import iterable_kind
from returns.interfaces.specific.future import FutureBased1
from returns.interfaces.specific.future_result import FutureResultBased2
from returns.io import IO, IOResult
from returns.primitives.container import BaseContainer
from returns.primitives.hkt import (
    Kind1,
    Kind2,
    SupportsKind1,
    SupportsKind2,
    dekind,
)
from returns.primitives.iterables import BaseIterableStrategyN, FailFast
from returns.primitives.reawaitable import ReAwaitable
from returns.result import Failure, Result, Success

# Definitions:
_ValueType = TypeVar('_ValueType', covariant=True)
_NewValueType = TypeVar('_NewValueType')
_ErrorType = TypeVar('_ErrorType', covariant=True)
_NewErrorType = TypeVar('_NewErrorType')

# Aliases:
_FirstType = TypeVar('_FirstType')
_SecondType = TypeVar('_SecondType')


# Public composition helpers:

[docs]async def async_identity(instance: _FirstType) -> _FirstType: """ Async function that returns its argument. .. code:: python >>> import anyio >>> from returns.future import async_identity >>> assert anyio.run(async_identity, 1) == 1 See :func:`returns.functions.identity` for sync version of this function and more docs and examples. """ return instance
# Future # ======
[docs]@final class Future( BaseContainer, SupportsKind1['Future', _ValueType], FutureBased1[_ValueType], ): """ Container to easily compose ``async`` functions. Represents a better abstraction over a simple coroutine. Is framework, event-loop, and IO-library agnostics. Works with ``asyncio``, ``curio``, ``trio``, or any other tool. Internally we use ``anyio`` to test that it works as expected for any io stack. Note that ``Future[a]`` represents a computation that never fails and returns ``IO[a]`` type. Use ``FutureResult[a, b]`` for operations that might fail. Like DB access or network operations. Is not related to ``asyncio.Future`` in any kind. .. rubric:: Tradeoffs Due to possible performance issues we move all coroutines definitions to a separate module. See also: https://gcanti.github.io/fp-ts/modules/Task.ts.html https://zio.dev/docs/overview/overview_basic_concurrency https://github.com/correl/typesafe-monads/blob/master/monads/future.py """ _inner_value: Awaitable[_ValueType] def __init__(self, inner_value: Awaitable[_ValueType]) -> None: """ Public constructor for this type. Also required for typing. .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> async def coro(arg: int) -> int: ... return arg + 1 >>> container = Future(coro(1)) >>> assert anyio.run(container.awaitable) == IO(2) """ super().__init__(ReAwaitable(inner_value)) def __await__(self) -> Generator[None, None, IO[_ValueType]]: """ By defining this magic method we make ``Future`` awaitable. This means you can use ``await`` keyword to evaluate this container: .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> async def main() -> IO[int]: ... return await Future.from_value(1) >>> assert anyio.run(main) == IO(1) When awaited we returned the value wrapped in :class:`returns.io.IO` container to indicate that the computation was impure. See also: https://docs.python.org/3/library/asyncio-task.html#awaitables https://www.python.org/dev/peps/pep-0492/#new-abstract-base-classes """ return self.awaitable().__await__() # noqa: WPS609
[docs] async def awaitable(self) -> IO[_ValueType]: """ Transforms ``Future[a]`` to ``Awaitable[IO[a]]``. Use this method when you need a real coroutine. Like for ``asyncio.run`` calls. Note, that returned value will be wrapped in :class:`returns.io.IO` container. .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> assert anyio.run(Future.from_value(1).awaitable) == IO(1) """ return IO(await self._inner_value)
[docs] def map( # noqa: WPS125 self, function: Callable[[_ValueType], _NewValueType], ) -> 'Future[_NewValueType]': """ Applies function to the inner value. Applies 'function' to the contents of the IO instance and returns a new ``Future`` object containing the result. 'function' should accept a single "normal" (non-container) argument and return a non-container result. .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> def mappable(x: int) -> int: ... return x + 1 >>> assert anyio.run( ... Future.from_value(1).map(mappable).awaitable, ... ) == IO(2) """ return Future(_future.async_map(function, self._inner_value))
[docs] def apply( self, container: Kind1['Future', Callable[[_ValueType], _NewValueType]], ) -> 'Future[_NewValueType]': """ Calls a wrapped function in a container on this container. .. code:: python >>> import anyio >>> from returns.future import Future >>> def transform(arg: int) -> str: ... return str(arg) + 'b' >>> assert anyio.run( ... Future.from_value(1).apply( ... Future.from_value(transform), ... ).awaitable, ... ) == IO('1b') """ return Future(_future.async_apply(dekind(container), self._inner_value))
[docs] def bind( self, function: Callable[[_ValueType], Kind1['Future', _NewValueType]], ) -> 'Future[_NewValueType]': """ Applies 'function' to the result of a previous calculation. 'function' should accept a single "normal" (non-container) argument and return ``Future`` type object. .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> def bindable(x: int) -> Future[int]: ... return Future.from_value(x + 1) >>> assert anyio.run( ... Future.from_value(1).bind(bindable).awaitable, ... ) == IO(2) """ return Future(_future.async_bind(function, self._inner_value))
#: Alias for `bind` method. Part of the `FutureBasedN` interface. bind_future = bind
[docs] def bind_async( self, function: Callable[ [_ValueType], Awaitable[Kind1['Future', _NewValueType]], ], ) -> 'Future[_NewValueType]': """ Compose a container and ``async`` function returning a container. This function should return a container value. See :meth:`~Future.bind_awaitable` to bind ``async`` function that returns a plain value. .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> async def coroutine(x: int) -> Future[str]: ... return Future.from_value(str(x + 1)) >>> assert anyio.run( ... Future.from_value(1).bind_async(coroutine).awaitable, ... ) == IO('2') """ return Future(_future.async_bind_async(function, self._inner_value))
#: Alias for `bind_async` method. Part of the `FutureBasedN` interface. bind_async_future = bind_async
[docs] def bind_awaitable( self, function: Callable[[_ValueType], 'Awaitable[_NewValueType]'], ) -> 'Future[_NewValueType]': """ Allows to compose a container and a regular ``async`` function. This function should return plain, non-container value. See :meth:`~Future.bind_async` to bind ``async`` function that returns a container. .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> async def coroutine(x: int) -> int: ... return x + 1 >>> assert anyio.run( ... Future.from_value(1).bind_awaitable(coroutine).awaitable, ... ) == IO(2) """ return Future(_future.async_bind_awaitable( function, self._inner_value, ))
[docs] def bind_io( self, function: Callable[[_ValueType], IO[_NewValueType]], ) -> 'Future[_NewValueType]': """ Applies 'function' to the result of a previous calculation. 'function' should accept a single "normal" (non-container) argument and return ``IO`` type object. .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> def bindable(x: int) -> IO[int]: ... return IO(x + 1) >>> assert anyio.run( ... Future.from_value(1).bind_io(bindable).awaitable, ... ) == IO(2) """ return Future(_future.async_bind_io(function, self._inner_value))
[docs] @classmethod def from_value(cls, inner_value: _NewValueType) -> 'Future[_NewValueType]': """ Allows to create a ``Future`` from a plain value. The resulting ``Future`` will just return the given value wrapped in :class:`returns.io.IO` container when awaited. .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> async def main() -> bool: ... return (await Future.from_value(1)) == IO(1) >>> assert anyio.run(main) is True """ return Future(async_identity(inner_value))
[docs] @classmethod def from_future( cls, inner_value: 'Future[_NewValueType]', ) -> 'Future[_NewValueType]': """ Creates a new ``Future`` from the existing one. .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> future = Future.from_value(1) >>> assert anyio.run(Future.from_future(future).awaitable) == IO(1) Part of the ``FutureBasedN`` interface. """ return inner_value
[docs] @classmethod def from_iterable( cls, inner_value: Iterable[Kind1['Future', _NewValueType]], strategy: Type[BaseIterableStrategyN] = FailFast, ) -> 'Future[Sequence[_NewValueType]]': """ Transforms an iterable of ``Future`` containers into a single container. .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> assert anyio.run(Future.from_iterable([ ... Future.from_value(1), ... Future.from_value(2), ... ]).awaitable) == IO((1, 2)) """ return iterable_kind(cls, inner_value, strategy)
[docs] @classmethod def from_io(cls, inner_value: IO[_NewValueType]) -> 'Future[_NewValueType]': """ Allows to create a ``Future`` from ``IO`` container. .. code:: python >>> import anyio >>> from returns.future import Future >>> from returns.io import IO >>> async def main() -> bool: ... return (await Future.from_io(IO(1))) == IO(1) >>> assert anyio.run(main) is True """ return Future(async_identity(inner_value._inner_value))
[docs] @classmethod def from_future_result( cls, inner_value: 'FutureResult[_NewValueType, _NewErrorType]', ) -> 'Future[Result[_NewValueType, _NewErrorType]]': """ Creates ``Future[Result[a, b]]`` instance from ``FutureResult[a, b]``. This method is the inverse of :meth:`~FutureResult.from_typecast`. .. code:: python >>> import anyio >>> from returns.future import Future, FutureResult >>> from returns.io import IO >>> from returns.result import Success >>> container = Future.from_future_result(FutureResult.from_value(1)) >>> assert anyio.run(container.awaitable) == IO(Success(1)) """ return Future(inner_value._inner_value)
# Decorators:
[docs]def future( function: Callable[ ..., Coroutine[_FirstType, _SecondType, _ValueType], ], ) -> Callable[..., Future[_ValueType]]: """ Decorator to turn a coroutine definition into ``Future`` container. .. code:: python >>> import anyio >>> from returns.io import IO >>> from returns.future import future >>> @future ... async def test(x: int) -> int: ... return x + 1 >>> assert anyio.run(test(1).awaitable) == IO(2) Requires our :ref:`mypy plugin <mypy-plugins>`. """ @wraps(function) def decorator(*args, **kwargs): return Future(function(*args, **kwargs)) return decorator
[docs]def asyncify(function: Callable[..., _ValueType]) -> Callable[ ..., Coroutine[Any, Any, _ValueType], ]: """ Decorator to turn a common function into an asynchronous function. This decorator is useful for composition with ``Future`` and ``FutureResult`` containers. .. warning:: This function will not your sync function **run** like async one. It will still be a blocking function that looks like async one. We recommend to only use this decorator with functions that do not access network or filesystem. It is only a composition helper, not a transformer. Usage example: .. code:: python >>> import anyio >>> from returns.future import asyncify >>> @asyncify ... def test(x: int) -> int: ... return x + 1 >>> assert anyio.run(test, 1) == 2 Requires our :ref:`mypy plugin <mypy-plugins>`. Read more about async and sync functions: https://journal.stuffwithstuff.com/2015/02/01/what-color-is-your-function/ """ @wraps(function) async def decorator(*args, **kwargs): return function(*args, **kwargs) return decorator
# FutureResult # ============
[docs]@final class FutureResult( BaseContainer, SupportsKind2['FutureResult', _ValueType, _ErrorType], FutureResultBased2[_ValueType, _ErrorType], ): """ Container to easily compose ``async`` functions. Represents a better abstraction over a simple coroutine. Is framework, event-loop, and IO-library agnostics. Works with ``asyncio``, ``curio``, ``trio``, or any other tool. Internally we use ``anyio`` to test that it works as expected for any io stack. Note that ``FutureResult[a, b]`` represents a computation that can fail and returns ``IOResult[a, b]`` type. Use ``Future[a]`` for operations that cannot fail. This is a ``Future`` that returns ``Result`` type. By providing this utility type we make developers' lifes easier. ``FutureResult`` has a lot of composition helpers to turn complex nested operations into a one function calls. .. rubric:: Tradeoffs Due to possible performance issues we move all coroutines definitions to a separate module. See also: https://gcanti.github.io/fp-ts/modules/TaskEither.ts.html https://zio.dev/docs/overview/overview_basic_concurrency """ _inner_value: Awaitable[Result[_ValueType, _ErrorType]] def __init__( self, inner_value: Awaitable[Result[_ValueType, _ErrorType]], ) -> None: """ Public constructor for this type. Also required for typing. .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess >>> from returns.result import Success, Result >>> async def coro(arg: int) -> Result[int, str]: ... return Success(arg + 1) >>> container = FutureResult(coro(1)) >>> assert anyio.run(container.awaitable) == IOSuccess(2) """ super().__init__(ReAwaitable(inner_value)) def __await__(self) -> Generator[ None, None, IOResult[_ValueType, _ErrorType], ]: """ By defining this magic method we make ``FutureResult`` awaitable. This means you can use ``await`` keyword to evaluate this container: .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess, IOResult >>> async def main() -> IOResult[int, str]: ... return await FutureResult.from_value(1) >>> assert anyio.run(main) == IOSuccess(1) When awaited we returned the value wrapped in :class:`returns.io.IOResult` container to indicate that the computation was impure and can fail. See also: https://docs.python.org/3/library/asyncio-task.html#awaitables https://www.python.org/dev/peps/pep-0492/#new-abstract-base-classes """ return self.awaitable().__await__() # noqa: WPS609
[docs] async def awaitable(self) -> IOResult[_ValueType, _ErrorType]: """ Transforms ``FutureResult[a, b]`` to ``Awaitable[IOResult[a, b]]``. Use this method when you need a real coroutine. Like for ``asyncio.run`` calls. Note, that returned value will be wrapped in :class:`returns.io.IOResult` container. .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess >>> assert anyio.run( ... FutureResult.from_value(1).awaitable, ... ) == IOSuccess(1) """ return IOResult.from_result(await self._inner_value)
[docs] def swap(self) -> 'FutureResult[_ErrorType, _ValueType]': """ Swaps value and error types. So, values become errors and errors become values. It is useful when you have to work with errors a lot. And since we have a lot of ``.bind_`` related methods and only a single ``.rescue``. It is easier to work with values than with errors. .. code:: python >>> import anyio >>> from returns.future import FutureSuccess, FutureFailure >>> from returns.io import IOSuccess, IOFailure >>> assert anyio.run(FutureSuccess(1).swap) == IOFailure(1) >>> assert anyio.run(FutureFailure(1).swap) == IOSuccess(1) """ return FutureResult(_future_result.async_swap(self._inner_value))
[docs] def map( # noqa: WPS125 self, function: Callable[[_ValueType], _NewValueType], ) -> 'FutureResult[_NewValueType, _ErrorType]': """ Applies function to the inner value. Applies 'function' to the contents of the IO instance and returns a new ``FutureResult`` object containing the result. 'function' should accept a single "normal" (non-container) argument and return a non-container result. .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess, IOFailure >>> def mappable(x: int) -> int: ... return x + 1 >>> assert anyio.run( ... FutureResult.from_value(1).map(mappable).awaitable, ... ) == IOSuccess(2) >>> assert anyio.run( ... FutureResult.from_failure(1).map(mappable).awaitable, ... ) == IOFailure(1) """ return FutureResult(_future_result.async_map( function, self._inner_value, ))
[docs] def apply( self, container: Kind2[ 'FutureResult', Callable[[_ValueType], _NewValueType], _ErrorType, ], ) -> 'FutureResult[_NewValueType, _ErrorType]': """ Calls a wrapped function in a container on this container. .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess, IOFailure >>> def appliable(x: int) -> int: ... return x + 1 >>> assert anyio.run( ... FutureResult.from_value(1).apply( ... FutureResult.from_value(appliable), ... ).awaitable, ... ) == IOSuccess(2) >>> assert anyio.run( ... FutureResult.from_failure(1).apply( ... FutureResult.from_value(appliable), ... ).awaitable, ... ) == IOFailure(1) >>> assert isinstance(anyio.run( ... FutureResult.from_value(1).apply( ... FutureResult.from_failure(appliable), ... ).awaitable, ... ), IOResult.failure_type) """ return FutureResult(_future_result.async_apply( dekind(container), self._inner_value, ))
[docs] def bind( self, function: Callable[ [_ValueType], Kind2['FutureResult', _NewValueType, _ErrorType], ], ) -> 'FutureResult[_NewValueType, _ErrorType]': """ Applies 'function' to the result of a previous calculation. 'function' should accept a single "normal" (non-container) argument and return ``Future`` type object. .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess, IOFailure >>> def bindable(x: int) -> FutureResult[int, str]: ... return FutureResult.from_value(x + 1) >>> assert anyio.run( ... FutureResult.from_value(1).bind(bindable).awaitable, ... ) == IOSuccess(2) >>> assert anyio.run( ... FutureResult.from_failure(1).bind(bindable).awaitable, ... ) == IOFailure(1) """ return FutureResult(_future_result.async_bind( function, self._inner_value, ))
#: Alias for `bind` method. #: Part of the `FutureResultBasedN` interface. bind_future_result = bind
[docs] def bind_async( self, function: Callable[ [_ValueType], Awaitable[Kind2['FutureResult', _NewValueType, _ErrorType]], ], ) -> 'FutureResult[_NewValueType, _ErrorType]': """ Composes a container and ``async`` function returning container. This function should return a container value. See :meth:`~FutureResult.bind_awaitable` to bind ``async`` function that returns a plain value. .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess, IOFailure >>> async def coroutine(x: int) -> FutureResult[str, int]: ... return FutureResult.from_value(str(x + 1)) >>> assert anyio.run( ... FutureResult.from_value(1).bind_async(coroutine).awaitable, ... ) == IOSuccess('2') >>> assert anyio.run( ... FutureResult.from_failure(1).bind_async(coroutine).awaitable, ... ) == IOFailure(1) """ return FutureResult(_future_result.async_bind_async( function, self._inner_value, ))
#: Alias for `bind_async` method. #: Part of the `FutureResultBasedN` interface. bind_async_future_result = bind_async
[docs] def bind_awaitable( self, function: Callable[[_ValueType], Awaitable[_NewValueType]], ) -> 'FutureResult[_NewValueType, _ErrorType]': """ Allows to compose a container and a regular ``async`` function. This function should return plain, non-container value. See :meth:`~FutureResult.bind_async` to bind ``async`` function that returns a container. .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess, IOFailure >>> async def coro(x: int) -> int: ... return x + 1 >>> assert anyio.run( ... FutureResult.from_value(1).bind_awaitable(coro).awaitable, ... ) == IOSuccess(2) >>> assert anyio.run( ... FutureResult.from_failure(1).bind_awaitable(coro).awaitable, ... ) == IOFailure(1) """ return FutureResult(_future_result.async_bind_awaitable( function, self._inner_value, ))
[docs] def bind_result( self, function: Callable[[_ValueType], Result[_NewValueType, _ErrorType]], ) -> 'FutureResult[_NewValueType, _ErrorType]': """ Binds a function returning ``Result[a, b]`` container. .. code:: python >>> import anyio >>> from returns.io import IOSuccess, IOFailure >>> from returns.result import Result, Success >>> from returns.future import FutureResult >>> def bind(inner_value: int) -> Result[int, str]: ... return Success(inner_value + 1) >>> assert anyio.run( ... FutureResult.from_value(1).bind_result(bind).awaitable, ... ) == IOSuccess(2) >>> assert anyio.run( ... FutureResult.from_failure('a').bind_result(bind).awaitable, ... ) == IOFailure('a') """ return FutureResult(_future_result.async_bind_result( function, self._inner_value, ))
[docs] def bind_ioresult( self, function: Callable[[_ValueType], IOResult[_NewValueType, _ErrorType]], ) -> 'FutureResult[_NewValueType, _ErrorType]': """ Binds a function returning ``IOResult[a, b]`` container. .. code:: python >>> import anyio >>> from returns.io import IOResult, IOSuccess, IOFailure >>> from returns.future import FutureResult >>> def bind(inner_value: int) -> IOResult[int, str]: ... return IOSuccess(inner_value + 1) >>> assert anyio.run( ... FutureResult.from_value(1).bind_ioresult(bind).awaitable, ... ) == IOSuccess(2) >>> assert anyio.run( ... FutureResult.from_failure('a').bind_ioresult(bind).awaitable, ... ) == IOFailure('a') """ return FutureResult(_future_result.async_bind_ioresult( function, self._inner_value, ))
[docs] def bind_io( self, function: Callable[[_ValueType], IO[_NewValueType]], ) -> 'FutureResult[_NewValueType, _ErrorType]': """ Binds a function returning ``IO[a]`` container. .. code:: python >>> import anyio >>> from returns.io import IO, IOSuccess, IOFailure >>> from returns.future import FutureResult >>> def bind(inner_value: int) -> IO[float]: ... return IO(inner_value + 0.5) >>> assert anyio.run( ... FutureResult.from_value(1).bind_io(bind).awaitable, ... ) == IOSuccess(1.5) >>> assert anyio.run( ... FutureResult.from_failure(1).bind_io(bind).awaitable, ... ) == IOFailure(1) """ return FutureResult(_future_result.async_bind_io( function, self._inner_value, ))
[docs] def bind_future( self, function: Callable[[_ValueType], Future[_NewValueType]], ) -> 'FutureResult[_NewValueType, _ErrorType]': """ Binds a function returning ``Future[a]`` container. .. code:: python >>> import anyio >>> from returns.io import IOSuccess, IOFailure >>> from returns.future import Future, FutureResult >>> def bind(inner_value: int) -> Future[float]: ... return Future.from_value(inner_value + 0.5) >>> assert anyio.run( ... FutureResult.from_value(1).bind_future(bind).awaitable, ... ) == IOSuccess(1.5) >>> assert anyio.run( ... FutureResult.from_failure(1).bind_future(bind).awaitable, ... ) == IOFailure(1) """ return FutureResult(_future_result.async_bind_future( function, self._inner_value, ))
[docs] def bind_async_future( self, function: Callable[[_ValueType], Awaitable['Future[_NewValueType]']], ) -> 'FutureResult[_NewValueType, _ErrorType]': """ Composes a container and ``async`` function returning ``Future``. Similar to :meth:`~FutureResult.bind_future` but works with async functions. .. code:: python >>> import anyio >>> from returns.future import Future, FutureResult >>> from returns.io import IOSuccess, IOFailure >>> async def coroutine(x: int) -> Future[str]: ... return Future.from_value(str(x + 1)) >>> assert anyio.run( ... FutureResult.from_value(1).bind_async_future, ... coroutine, ... ) == IOSuccess('2') >>> assert anyio.run( ... FutureResult.from_failure(1).bind_async, ... coroutine, ... ) == IOFailure(1) """ return FutureResult(_future_result.async_bind_async_future( function, self._inner_value, ))
[docs] def alt( self, function: Callable[[_ErrorType], _NewErrorType], ) -> 'FutureResult[_ValueType, _NewErrorType]': """ Composes failed container with a pure function to modify failure. .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess, IOFailure >>> def altable(arg: int) -> int: ... return arg + 1 >>> assert anyio.run( ... FutureResult.from_value(1).alt(altable).awaitable, ... ) == IOSuccess(1) >>> assert anyio.run( ... FutureResult.from_failure(1).alt(altable).awaitable, ... ) == IOFailure(2) """ return FutureResult(_future_result.async_alt( function, self._inner_value, ))
[docs] def rescue( self, function: Callable[ [_ErrorType], Kind2['FutureResult', _ValueType, _NewErrorType], ], ) -> 'FutureResult[_ValueType, _NewErrorType]': """ Composes failed container with a function that returns a container. .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess >>> def rescuable(x: int) -> FutureResult[int, str]: ... return FutureResult.from_value(x + 1) >>> assert anyio.run( ... FutureResult.from_value(1).rescue(rescuable).awaitable, ... ) == IOSuccess(1) >>> assert anyio.run( ... FutureResult.from_failure(1).rescue(rescuable).awaitable, ... ) == IOSuccess(2) """ return FutureResult(_future_result.async_rescue( function, self._inner_value, ))
[docs] def compose_result( self, function: Callable[ [Result[_ValueType, _ErrorType]], Kind2['FutureResult', _NewValueType, _ErrorType], ], ) -> 'FutureResult[_NewValueType, _ErrorType]': """ Composes inner ``Result`` with ``FutureResult`` returning function. Can be useful when you need an access to both states of the result. .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess, IOFailure >>> from returns.result import Result >>> def count(container: Result[int, int]) -> FutureResult[int, int]: ... return FutureResult.from_result( ... container.map(lambda x: x + 1).alt(abs), ... ) >>> assert anyio.run( ... FutureResult.from_value(1).compose_result, count, ... ) == IOSuccess(2) >>> assert anyio.run( ... FutureResult.from_failure(-1).compose_result, count, ... ) == IOFailure(1) """ return FutureResult(_future_result.async_compose_result( function, self._inner_value, ))
[docs] @classmethod def from_typecast( cls, inner_value: Future[Result[_NewValueType, _NewErrorType]], ) -> 'FutureResult[_NewValueType, _NewErrorType]': """ Creates ``FutureResult[a, b]`` from ``Future[Result[a, b]]``. .. code:: python >>> import anyio >>> from returns.io import IOSuccess, IOFailure >>> from returns.result import Success, Failure >>> from returns.future import Future, FutureResult >>> async def main(): ... assert await FutureResult.from_typecast( ... Future.from_value(Success(1)), ... ) == IOSuccess(1) ... assert await FutureResult.from_typecast( ... Future.from_value(Failure(1)), ... ) == IOFailure(1) >>> anyio.run(main) """ return FutureResult(inner_value._inner_value)
[docs] @classmethod def from_future( cls, inner_value: Future[_NewValueType], ) -> 'FutureResult[_NewValueType, Any]': """ Creates ``FutureResult`` from successful ``Future`` value. .. code:: python >>> import anyio >>> from returns.io import IOSuccess >>> from returns.future import Future, FutureResult >>> async def main(): ... assert await FutureResult.from_future( ... Future.from_value(1), ... ) == IOSuccess(1) >>> anyio.run(main) """ return FutureResult(_future_result.async_from_success(inner_value))
[docs] @classmethod def from_failed_future( cls, inner_value: Future[_NewErrorType], ) -> 'FutureResult[Any, _NewErrorType]': """ Creates ``FutureResult`` from failed ``Future`` value. .. code:: python >>> import anyio >>> from returns.io import IOFailure >>> from returns.future import Future, FutureResult >>> async def main(): ... assert await FutureResult.from_failed_future( ... Future.from_value(1), ... ) == IOFailure(1) >>> anyio.run(main) """ return FutureResult(_future_result.async_from_failure(inner_value))
[docs] @classmethod def from_future_result( cls, inner_value: 'FutureResult[_NewValueType, _NewErrorType]', ) -> 'FutureResult[_NewValueType, _NewErrorType]': """ Creates new ``FutureResult`` from existing one. .. code:: python >>> import anyio >>> from returns.io import IOSuccess >>> from returns.future import FutureResult >>> async def main(): ... assert await FutureResult.from_future_result( ... FutureResult.from_value(1), ... ) == IOSuccess(1) >>> anyio.run(main) Part of the ``FutureResultLikeN`` interface. """ return inner_value
[docs] @classmethod def from_io( cls, inner_value: IO[_NewValueType], ) -> 'FutureResult[_NewValueType, Any]': """ Creates ``FutureResult`` from successful ``IO`` value. .. code:: python >>> import anyio >>> from returns.io import IO, IOSuccess >>> from returns.future import FutureResult >>> async def main(): ... assert await FutureResult.from_io( ... IO(1), ... ) == IOSuccess(1) >>> anyio.run(main) """ return FutureResult.from_value(inner_value._inner_value)
[docs] @classmethod def from_failed_io( cls, inner_value: IO[_NewErrorType], ) -> 'FutureResult[Any, _NewErrorType]': """ Creates ``FutureResult`` from failed ``IO`` value. .. code:: python >>> import anyio >>> from returns.io import IO, IOFailure >>> from returns.future import FutureResult >>> async def main(): ... assert await FutureResult.from_failed_io( ... IO(1), ... ) == IOFailure(1) >>> anyio.run(main) """ return FutureResult.from_failure(inner_value._inner_value)
[docs] @classmethod def from_ioresult( cls, inner_value: IOResult[_NewValueType, _NewErrorType], ) -> 'FutureResult[_NewValueType, _NewErrorType]': """ Creates ``FutureResult`` from ``IOResult`` value. .. code:: python >>> import anyio >>> from returns.io import IOSuccess, IOFailure >>> from returns.future import FutureResult >>> async def main(): ... assert await FutureResult.from_ioresult( ... IOSuccess(1), ... ) == IOSuccess(1) ... assert await FutureResult.from_ioresult( ... IOFailure(1), ... ) == IOFailure(1) >>> anyio.run(main) """ return FutureResult(async_identity(inner_value._inner_value))
[docs] @classmethod def from_result( cls, inner_value: Result[_NewValueType, _NewErrorType], ) -> 'FutureResult[_NewValueType, _NewErrorType]': """ Creates ``FutureResult`` from ``Result`` value. .. code:: python >>> import anyio >>> from returns.io import IOSuccess, IOFailure >>> from returns.result import Success, Failure >>> from returns.future import FutureResult >>> async def main(): ... assert await FutureResult.from_result( ... Success(1), ... ) == IOSuccess(1) ... assert await FutureResult.from_result( ... Failure(1), ... ) == IOFailure(1) >>> anyio.run(main) """ return FutureResult(async_identity(inner_value))
[docs] @classmethod def from_value( cls, inner_value: _NewValueType, ) -> 'FutureResult[_NewValueType, Any]': """ Creates ``FutureResult`` from successful value. .. code:: python >>> import anyio >>> from returns.io import IOSuccess >>> from returns.future import FutureResult >>> async def main(): ... assert await FutureResult.from_value( ... 1, ... ) == IOSuccess(1) >>> anyio.run(main) """ return FutureResult(async_identity(Success(inner_value)))
[docs] @classmethod def from_failure( cls, inner_value: _NewErrorType, ) -> 'FutureResult[Any, _NewErrorType]': """ Creates ``FutureResult`` from failed value. .. code:: python >>> import anyio >>> from returns.io import IOFailure >>> from returns.future import FutureResult >>> async def main(): ... assert await FutureResult.from_failure( ... 1, ... ) == IOFailure(1) >>> anyio.run(main) """ return FutureResult(async_identity(Failure(inner_value)))
[docs] @classmethod def from_iterable( cls, inner_value: Iterable[ Kind2['FutureResult', _NewValueType, _NewErrorType], ], strategy: Type[BaseIterableStrategyN] = FailFast, ) -> 'FutureResult[Sequence[_NewValueType], _NewErrorType]': """ Transforms an iterable of ``FutureResult`` containers. Returns a single container with multiple elements inside. .. code:: python >>> import anyio >>> from returns.future import FutureResult >>> from returns.io import IOSuccess >>> all_success = FutureResult.from_iterable([ ... FutureResult.from_value(1), ... FutureResult.from_value(2), ... ]) >>> assert anyio.run(all_success.awaitable) == IOSuccess((1, 2)) """ return iterable_kind(cls, inner_value, strategy)
[docs]def FutureSuccess( # noqa: N802 inner_value: _NewValueType, ) -> FutureResult[_NewValueType, Any]: """ Public unit function to create successful ``FutureResult`` objects. Is the same as :meth:`~FutureResult.from_value`. .. code:: python >>> import anyio >>> from returns.future import FutureResult, FutureSuccess >>> assert anyio.run(FutureSuccess(1).awaitable) == anyio.run( ... FutureResult.from_value(1).awaitable, ... ) """ return FutureResult.from_value(inner_value)
[docs]def FutureFailure( # noqa: N802 inner_value: _NewErrorType, ) -> FutureResult[Any, _NewErrorType]: """ Public unit function to create failed ``FutureResult`` objects. Is the same as :meth:`~FutureResult.from_failure`. .. code:: python >>> import anyio >>> from returns.future import FutureResult, FutureFailure >>> assert anyio.run(FutureFailure(1).awaitable) == anyio.run( ... FutureResult.from_failure(1).awaitable, ... ) """ return FutureResult.from_failure(inner_value)
# Aliases: #: Alias for a popular case when ``Result`` has ``Exception`` as error type. FutureResultE = FutureResult[_ValueType, Exception] # Decorators:
[docs]def future_safe( function: Callable[..., Coroutine[_FirstType, _SecondType, _ValueType]], ) -> Callable[..., FutureResultE[_ValueType]]: """ Decorator to convert exception-throwing coroutine to ``FutureResult``. Should be used with care, since it only catches ``Exception`` subclasses. It does not catch ``BaseException`` subclasses. If you need to mark sync function as ``safe``, use :func:`returns.future.future_safe` instead. This decorator only works with ``async`` functions. Example: .. code:: python >>> import anyio >>> from returns.future import future_safe >>> from returns.io import IOSuccess, IOResult >>> @future_safe ... async def might_raise(arg: int) -> float: ... return 1 / arg ... >>> assert anyio.run(might_raise(2).awaitable) == IOSuccess(0.5) >>> assert isinstance( ... anyio.run(might_raise(0).awaitable), ... IOResult.failure_type, ... ) Similar to :func:`returns.io.impure_safe` and :func:`returns.result.safe` decorators, but works with ``async`` functions. Requires our :ref:`mypy plugin <mypy-plugins>`. """ async def factory(*args, **kwargs) -> Result[_ValueType, Exception]: try: return Success(await function(*args, **kwargs)) except Exception as exc: return Failure(exc) @wraps(function) def decorator(*args, **kwargs): return FutureResult(factory(*args, **kwargs)) return decorator