Future

A set of primitives to work with async functions.

Can be used with asyncio, trio, and curio. And any event-loop!

Tested with anyio.

What problems do we solve with these containers? Basically these ones:

  1. You cannot call async function from a sync one

  2. Any unexpectedly thrown exception can ruin your whole event loop

  3. Ugly composition with lots of await statements

Future container

Without Future container it is impossible to compose two functions: sync and async one.

You simply cannot await coroutines inside a sync context. It is even a SyntaxError.

def test():
    await some()
# SyntaxError: 'await' outside async function

So, you have to turn you function into async one. And all callers of this function in async functions. And all their callers.

This is really hard to model. When you code has two types of uncomposable functions you increase your mental complexity by extreme levels.

Instead, you can use Future container, it allows you to model async interactions in a sync manner:

>>> from returns.future import Future

>>> async def first() -> int:
...     return 1

>>> async def second(arg: int) -> int:
...     return arg + 1

>>> def main() -> Future[int]:  # sync function!
...    return Future(first()).bind_awaitable(second)

Now we can compose async functions and maintaining a sync context! It is also possible run a Future with regular tools like asyncio.run or anyio.run:

>>> import anyio
>>> from returns.io import IO

>>> assert anyio.run(main().awaitable) == IO(2)

One more very useful thing Future does behind the scenes is converting its result to IO-based containers. This helps a lot when separating pure and impure (async functions are impure) code inside your app.

FutureResult

This container becomes very useful when working with async function that can fail.

It works the similar way regular Result does. And is literally a wrapper around Future[Result[_V, _E]] type.

Let’s see how it can be used in a real program:

 1import asyncio  # we use `asyncio` only as an example, you can use any io lib
 2from typing import Sequence, cast
 3
 4import httpx  # you would need to `pip install httpx`
 5from typing_extensions import Final, TypedDict
 6
 7from returns.future import FutureResultE, future_safe
 8from returns.io import IOResultE
 9from returns.iterables import Fold
10
11_URL: Final = 'https://jsonplaceholder.typicode.com/posts/{0}'
12_Post = TypedDict('_Post', {
13    'id': int,
14    'userId': int,
15    'title': str,
16    'body': str,
17})
18
19
20@future_safe
21async def _fetch_post(post_id: int) -> _Post:
22    # Ideally, we can use `ReaderFutureResult` to provide `client` from deps.
23    async with httpx.AsyncClient(timeout=5) as client:
24        response = await client.get(_URL.format(post_id))
25        response.raise_for_status()
26        return cast(_Post, response.json())  # or validate the response
27
28
29def _show_titles(number_of_posts: int) -> Sequence[FutureResultE[str]]:
30    def factory(post: _Post) -> str:
31        return post['title']
32
33    return [
34        # Notice how easily we compose async and sync functions:
35        _fetch_post(post_id).map(factory)
36        # TODO: try `for post_id in (2, 1, 0):` to see how async errors work
37        for post_id in range(1, number_of_posts + 1)
38    ]
39
40
41async def main() -> IOResultE[Sequence[str]]:
42    """
43    Main entrypoint for the async world.
44
45    Let's fetch 3 titles of posts asynchronously.
46    We use `gather` to run requests in "parallel".
47    """
48    futures: Sequence[IOResultE[str]] = await asyncio.gather(*_show_titles(3))
49    return Fold.collect(futures, IOResultE.from_value(()))
50
51
52if __name__ == '__main__':
53    print(asyncio.run(main()))  # noqa: WPS421
54    # <IOResult: <Success: (
55    #    'sunt aut facere repellat provident occaecati ...',
56    #    'qui est esse',
57    #    'ea molestias quasi exercitationem repellat qui ipsa sit aut',
58    # )>>

What is different?

  1. We can now easily make show_titles sync, we can also make _fetch_post sync, but we would need to use ReaderFutureResult container with proper dependencies in this case

  2. We now don’t care about errors at all. In this example any error will cancel the whole pipeline

  3. We now have .map method to easily compose sync and async functions

You can see the next example with RequiresContextFutureResult and without a single async/await. That example illustrates the whole point of our actions: writing sync code that executes asynchronously without any magic at all.

Aliases

There are several useful alises for FutureResult type with some common values:

  • returns.future.FutureResultE is an alias for FutureResult[... Exception], just use it when you want to work with FutureResult containers that use exceptions as error type. It is named FutureResultE because it is FutureResultException and FutureResultError at the same time.

Decorators

future

This decorator helps to easily transform async def into Future:

>>> import anyio
>>> from returns.future import future, Future
>>> from returns.io import IO

>>> @future
... async def test(arg: int) -> float:
...     return arg / 2

>>> future_instance = test(1)
>>> assert isinstance(future_instance, Future)
>>> assert anyio.run(future_instance.awaitable) == IO(0.5)

Make sure that you decorate with @future only coroutines that do not throw exceptions. For ones that do, use future_safe.

future_safe

This decorator converts async def into FutureResult, which means that it becomes:

  1. Full featured Future like container

  2. Safe from any exceptions

Let’s dig into it:

>>> import anyio
>>> from returns.future import future_safe, FutureResult
>>> from returns.io import IOSuccess, IOFailure

>>> @future_safe
... async def test(arg: int) -> float:
...     return 1 / arg

>>> future_instance = test(2)
>>> assert isinstance(future_instance, FutureResult)
>>> assert anyio.run(future_instance.awaitable) == IOSuccess(0.5)

>>> str(anyio.run(test(0).awaitable))  # this will fail
'<IOResult: <Failure: division by zero>>'

Never miss exceptions ever again!

asyncify

Helper decorator to transform regular sync function into async ones.

>>> import anyio
>>> from inspect import iscoroutinefunction
>>> from returns.future import asyncify

>>> @asyncify
... def your_function(x: int) -> int:
...     return x + 1

>>> assert iscoroutinefunction(your_function) is True
>>> assert anyio.run(your_function, 1) == 2

Very important node: making your function async does not mean it will work asynchronously. It can still block if it uses blocking calls. Here’s an example of how you must not do:

import requests
from returns.future import asyncify

@asyncify
def please_do_not_do_that():
    return requests.get('...')  # this will still block!

Do not overuse this decorator. It is only useful for some basic composition with Future and FutureResult.

FAQ

How to create unit objects?

For Future container:

  • from_value when you have a raw value

  • from_io when you have existing IO container

  • from_future_result when you have existing FutureResult

For FutureResult container:

  • from_value when you want to mark some raw value as a Success

  • from_failure when you want to mark some raw value as a Failure

  • from_result when you already have Result container

  • from_io when you have successful IO

  • from_failed_io when you have failed IO

  • from_future when you have successful Future

  • from_failed_future when you have failed Future

  • from_typecast when you have existing Future[Result]

What is the difference between Future[Result[a, b]] and FutureResult[a, b]?

There’s almost none.

The only difference is that FutureResult[a, b] is a handy wrapper around Future[Result[a, b]], so you won’t need to use methods like .map and .bind twice.

You can always convert it with methods like .from_typecast and .from_future_result.

Further reading

API Reference

classDiagram FutureBasedN <|-- Future BaseContainer <|-- FutureResult SupportsKindN <|-- Future BaseContainer <|-- Future SupportsKindN <|-- FutureResult FutureResultBasedN <|-- FutureResult
async async_identity(instance)[source]

Async function that returns its argument.

>>> import anyio
>>> from returns.future import async_identity
>>> assert anyio.run(async_identity, 1) == 1

See returns.functions.identity() for sync version of this function and more docs and examples.

Parameters

instance (~_FirstType) –

Return type

~_FirstType

class Future(inner_value)[source]

Bases: returns.primitives.container.BaseContainer, returns.primitives.hkt.SupportsKindN[Future, returns.future._ValueType, NoReturn, NoReturn], returns.interfaces.specific.future.FutureBasedN[returns.future._ValueType, NoReturn, NoReturn]

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.

Tradeoffs

Due to possible performance issues we move all coroutines definitions to a separate module.

Parameters

inner_value (Awaitable[+_ValueType]) –

async awaitable()[source]

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 returns.io.IO container.

>>> import anyio
>>> from returns.future import Future
>>> from returns.io import IO
>>> assert anyio.run(Future.from_value(1).awaitable) == IO(1)
Return type

IO[+_ValueType]

map(function)[source]

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.

>>> 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)
Parameters

function (Callable[[+_ValueType], ~_NewValueType]) –

Return type

Future[~_NewValueType]

apply(container)[source]

Calls a wrapped function in a container on this container.

>>> 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')
Parameters

container (KindN[Future, Callable[[+_ValueType], ~_NewValueType], Any, Any]) –

Return type

Future[~_NewValueType]

bind(function)[source]

Applies ‘function’ to the result of a previous calculation.

‘function’ should accept a single “normal” (non-container) argument and return Future type object.

>>> 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)
Parameters

function (Callable[[+_ValueType], KindN[Future, ~_NewValueType, Any, Any]]) –

Return type

Future[~_NewValueType]

bind_future(function)

Alias for bind method. Part of the FutureBasedN interface.

Parameters

function (Callable[[+_ValueType], KindN[Future, ~_NewValueType, Any, Any]]) –

Return type

Future[~_NewValueType]

bind_async(function)[source]

Compose a container and async function returning a container.

This function should return a container value. See bind_awaitable() to bind async function that returns a plain value.

>>> 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')
Parameters

function (Callable[[+_ValueType], Awaitable[KindN[Future, ~_NewValueType, Any, Any]]]) –

Return type

Future[~_NewValueType]

bind_async_future(function)

Alias for bind_async method. Part of the FutureBasedN interface.

Parameters

function (Callable[[+_ValueType], Awaitable[KindN[Future, ~_NewValueType, Any, Any]]]) –

Return type

Future[~_NewValueType]

bind_awaitable(function)[source]

Allows to compose a container and a regular async function.

This function should return plain, non-container value. See bind_async() to bind async function that returns a container.

>>> 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)
Parameters

function (Callable[[+_ValueType], Awaitable[~_NewValueType]]) –

Return type

Future[~_NewValueType]

bind_io(function)[source]

Applies ‘function’ to the result of a previous calculation.

‘function’ should accept a single “normal” (non-container) argument and return IO type object.

>>> 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)
Parameters

function (Callable[[+_ValueType], IO[~_NewValueType]]) –

Return type

Future[~_NewValueType]

classmethod from_value(inner_value)[source]

Allows to create a Future from a plain value.

The resulting Future will just return the given value wrapped in returns.io.IO container when awaited.

>>> 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
Parameters

inner_value (~_NewValueType) –

Return type

Future[~_NewValueType]

classmethod from_future(inner_value)[source]

Creates a new Future from the existing one.

>>> 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.

Parameters

inner_value (Future[~_NewValueType]) –

Return type

Future[~_NewValueType]

classmethod from_io(inner_value)[source]

Allows to create a Future from IO container.

>>> 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
Parameters

inner_value (IO[~_NewValueType]) –

Return type

Future[~_NewValueType]

classmethod from_future_result(inner_value)[source]

Creates Future[Result[a, b]] instance from FutureResult[a, b].

This method is the inverse of from_typecast().

>>> 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))
Parameters

inner_value (FutureResult[~_NewValueType, ~_NewErrorType]) –

Return type

Future[Result[~_NewValueType, ~_NewErrorType]]

future(function)[source]

Decorator to turn a coroutine definition into Future container.

>>> 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 mypy plugin.

Parameters

function (Callable[…, Coroutine[~_FirstType, ~_SecondType, +_ValueType]]) –

Return type

Callable[…, Future[+_ValueType]]

asyncify(function)[source]

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:

>>> import anyio
>>> from returns.future import asyncify

>>> @asyncify
... def test(x: int) -> int:
...     return x + 1

>>> assert anyio.run(test, 1) == 2

Requires our mypy plugin.

Read more about async and sync functions: https://journal.stuffwithstuff.com/2015/02/01/what-color-is-your-function/

Parameters

function (Callable[…, +_ValueType]) –

Return type

Callable[…, Coroutine[Any, Any, +_ValueType]]

class FutureResult(inner_value)[source]

Bases: returns.primitives.container.BaseContainer, returns.primitives.hkt.SupportsKindN[FutureResult, returns.future._ValueType, returns.future._ErrorType, NoReturn], returns.interfaces.specific.future_result.FutureResultBasedN[returns.future._ValueType, returns.future._ErrorType, NoReturn]

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’ lives easier. FutureResult has a lot of composition helpers to turn complex nested operations into a one function calls.

Tradeoffs

Due to possible performance issues we move all coroutines definitions to a separate module.

Parameters

inner_value (Awaitable[Result[+_ValueType, +_ErrorType]]) –

async awaitable()[source]

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 returns.io.IOResult container.

>>> import anyio
>>> from returns.future import FutureResult
>>> from returns.io import IOSuccess
>>> assert anyio.run(
...     FutureResult.from_value(1).awaitable,
... ) == IOSuccess(1)
Return type

IOResult[+_ValueType, +_ErrorType]

swap()[source]

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 .lash. It is easier to work with values than with errors.

>>> 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 type

FutureResult[+_ErrorType, +_ValueType]

map(function)[source]

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.

>>> 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)
Parameters

function (Callable[[+_ValueType], ~_NewValueType]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

apply(container)[source]

Calls a wrapped function in a container on this container.

>>> 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 anyio.run(
...     FutureResult.from_value(1).apply(
...         FutureResult.from_failure(2),
...     ).awaitable,
... ) == IOFailure(2)
>>> assert anyio.run(
...     FutureResult.from_failure(1).apply(
...         FutureResult.from_failure(2),
...     ).awaitable,
... ) == IOFailure(1)
Parameters

container (KindN[FutureResult, Callable[[+_ValueType], ~_NewValueType], +_ErrorType, Any]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

bind(function)[source]

Applies ‘function’ to the result of a previous calculation.

‘function’ should accept a single “normal” (non-container) argument and return Future type object.

>>> 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)
Parameters

function (Callable[[+_ValueType], KindN[FutureResult, ~_NewValueType, +_ErrorType, Any]]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

bind_future_result(function)

Alias for bind method. Part of the FutureResultBasedN interface.

Parameters

function (Callable[[+_ValueType], KindN[FutureResult, ~_NewValueType, +_ErrorType, Any]]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

bind_async(function)[source]

Composes a container and async function returning container.

This function should return a container value. See bind_awaitable() to bind async function that returns a plain value.

>>> 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)
Parameters

function (Callable[[+_ValueType], Awaitable[KindN[FutureResult, ~_NewValueType, +_ErrorType, Any]]]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

bind_async_future_result(function)

Alias for bind_async method. Part of the FutureResultBasedN interface.

Parameters

function (Callable[[+_ValueType], Awaitable[KindN[FutureResult, ~_NewValueType, +_ErrorType, Any]]]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

bind_awaitable(function)[source]

Allows to compose a container and a regular async function.

This function should return plain, non-container value. See bind_async() to bind async function that returns a container.

>>> 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)
Parameters

function (Callable[[+_ValueType], Awaitable[~_NewValueType]]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

bind_result(function)[source]

Binds a function returning Result[a, b] container.

>>> 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')
Parameters

function (Callable[[+_ValueType], Result[~_NewValueType, +_ErrorType]]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

bind_ioresult(function)[source]

Binds a function returning IOResult[a, b] container.

>>> 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')
Parameters

function (Callable[[+_ValueType], IOResult[~_NewValueType, +_ErrorType]]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

bind_io(function)[source]

Binds a function returning IO[a] container.

>>> 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)
Parameters

function (Callable[[+_ValueType], IO[~_NewValueType]]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

bind_future(function)[source]

Binds a function returning Future[a] container.

>>> 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)
Parameters

function (Callable[[+_ValueType], Future[~_NewValueType]]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

bind_async_future(function)[source]

Composes a container and async function returning Future.

Similar to bind_future() but works with async functions.

>>> 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)
Parameters

function (Callable[[+_ValueType], Awaitable[Future[~_NewValueType]]]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

alt(function)[source]

Composes failed container with a pure function to modify failure.

>>> 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)
Parameters

function (Callable[[+_ErrorType], ~_NewErrorType]) –

Return type

FutureResult[+_ValueType, ~_NewErrorType]

lash(function)[source]

Composes failed container with a function that returns a container.

>>> import anyio
>>> from returns.future import FutureResult
>>> from returns.io import IOSuccess

>>> def lashable(x: int) -> FutureResult[int, str]:
...    return FutureResult.from_value(x + 1)

>>> assert anyio.run(
...     FutureResult.from_value(1).lash(lashable).awaitable,
... ) == IOSuccess(1)
>>> assert anyio.run(
...     FutureResult.from_failure(1).lash(lashable).awaitable,
... ) == IOSuccess(2)
Parameters

function (Callable[[+_ErrorType], KindN[FutureResult, +_ValueType, ~_NewErrorType, Any]]) –

Return type

FutureResult[+_ValueType, ~_NewErrorType]

compose_result(function)[source]

Composes inner Result with FutureResult returning function.

Can be useful when you need an access to both states of the result.

>>> 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)
Parameters

function (Callable[[Result[+_ValueType, +_ErrorType]], KindN[FutureResult, ~_NewValueType, +_ErrorType, Any]]) –

Return type

FutureResult[~_NewValueType, +_ErrorType]

classmethod from_typecast(inner_value)[source]

Creates FutureResult[a, b] from Future[Result[a, b]].

>>> 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)
Parameters

inner_value (Future[Result[~_NewValueType, ~_NewErrorType]]) –

Return type

FutureResult[~_NewValueType, ~_NewErrorType]

classmethod from_future(inner_value)[source]

Creates FutureResult from successful Future value.

>>> 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)
Parameters

inner_value (Future[~_NewValueType]) –

Return type

FutureResult[~_NewValueType, Any]

classmethod from_failed_future(inner_value)[source]

Creates FutureResult from failed Future value.

>>> 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)
Parameters

inner_value (Future[~_NewErrorType]) –

Return type

FutureResult[Any, ~_NewErrorType]

classmethod from_future_result(inner_value)[source]

Creates new FutureResult from existing one.

>>> 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.

Parameters

inner_value (FutureResult[~_NewValueType, ~_NewErrorType]) –

Return type

FutureResult[~_NewValueType, ~_NewErrorType]

classmethod from_io(inner_value)[source]

Creates FutureResult from successful IO value.

>>> 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)
Parameters

inner_value (IO[~_NewValueType]) –

Return type

FutureResult[~_NewValueType, Any]

classmethod from_failed_io(inner_value)[source]

Creates FutureResult from failed IO value.

>>> 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)
Parameters

inner_value (IO[~_NewErrorType]) –

Return type

FutureResult[Any, ~_NewErrorType]

classmethod from_ioresult(inner_value)[source]

Creates FutureResult from IOResult value.

>>> 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)
Parameters

inner_value (IOResult[~_NewValueType, ~_NewErrorType]) –

Return type

FutureResult[~_NewValueType, ~_NewErrorType]

classmethod from_result(inner_value)[source]

Creates FutureResult from Result value.

>>> 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)
Parameters

inner_value (Result[~_NewValueType, ~_NewErrorType]) –

Return type

FutureResult[~_NewValueType, ~_NewErrorType]

classmethod from_value(inner_value)[source]

Creates FutureResult from successful value.

>>> 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)
Parameters

inner_value (~_NewValueType) –

Return type

FutureResult[~_NewValueType, Any]

classmethod from_failure(inner_value)[source]

Creates FutureResult from failed value.

>>> 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)
Parameters

inner_value (~_NewErrorType) –

Return type

FutureResult[Any, ~_NewErrorType]

FutureSuccess(inner_value)[source]

Public unit function to create successful FutureResult objects.

Is the same as from_value().

>>> import anyio
>>> from returns.future import FutureResult, FutureSuccess

>>> assert anyio.run(FutureSuccess(1).awaitable) == anyio.run(
...     FutureResult.from_value(1).awaitable,
... )
Parameters

inner_value (~_NewValueType) –

Return type

FutureResult[~_NewValueType, Any]

FutureFailure(inner_value)[source]

Public unit function to create failed FutureResult objects.

Is the same as from_failure().

>>> import anyio
>>> from returns.future import FutureResult, FutureFailure

>>> assert anyio.run(FutureFailure(1).awaitable) == anyio.run(
...     FutureResult.from_failure(1).awaitable,
... )
Parameters

inner_value (~_NewErrorType) –

Return type

FutureResult[Any, ~_NewErrorType]

FutureResultE

Alias for a popular case when Result has Exception as error type.

alias of returns.future.FutureResult[returns.future._ValueType, Exception]

future_safe(function)[source]

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 returns.future.future_safe() instead. This decorator only works with async functions. Example:

>>> 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 returns.io.impure_safe() and returns.result.safe() decorators, but works with async functions.

Requires our mypy plugin.

Parameters

function (Callable[…, Coroutine[~_FirstType, ~_SecondType, +_ValueType]]) –

Return type

Callable[…, FutureResult[+_ValueType, Exception]]

IO Context