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 your 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 to 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:

import asyncio  # we use `asyncio` only as an example, you can use any io lib
from collections.abc import Sequence
from typing import Final, cast

import httpx  # you would need to `pip install httpx`
from typing_extensions import TypedDict

from returns.future import FutureResult, future_safe
from returns.io import IOResultE
from returns.iterables import Fold

_URL: Final = 'https://jsonplaceholder.typicode.com/posts/{0}'


class _Post(TypedDict):
    id: int
    user_id: int
    title: str
    body: str


@future_safe
async def _fetch_post(post_id: int) -> _Post:
    # Ideally, we can use `ReaderFutureResult` to provide `client` from deps.
    async with httpx.AsyncClient(timeout=5) as client:
        response = await client.get(_URL.format(post_id))
        response.raise_for_status()
        return cast(_Post, response.json())  # or validate the response


def _show_titles(
    number_of_posts: int,
) -> Sequence[FutureResult[str, Exception]]:
    def factory(post: _Post) -> str:  # noqa: FURB118
        return post['title']

    return [
        # Notice how easily we compose async and sync functions:
        _fetch_post(post_id).map(factory)
        # TODO: try `for post_id in (2, 1, 0):` to see how async errors work
        for post_id in range(1, number_of_posts + 1)
    ]


async def main() -> IOResultE[Sequence[str]]:
    """
    Main entrypoint for the async world.

    Let's fetch 3 titles of posts asynchronously.
    We use `gather` to run requests in "parallel".
    """
    futures: Sequence[IOResultE[str]] = await asyncio.gather(*_show_titles(3))
    return Fold.collect(futures, IOResultE.from_value(()))


if __name__ == '__main__':
    print(asyncio.run(main()))  # noqa: WPS421
    # <IOResult: <Success: (
    #    'sunt aut facere repellat provident occaecati ...',
    #    'qui est esse',
    #    'ea molestias quasi exercitationem repellat qui ipsa sit aut',
    # )>>

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.

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

Is it somehow related to Future object from asyncio?

Nope, we just use the same naming there are in other languages and platforms. Python happens to have its own meaning for this word.

In our worldview, these two Future entities should never meet each other in a single codebase.

It is also not related to concurrent.Future.

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

• How Async Should Have Been

• What Color is Your Function?

• From Promises to Futures

API Reference

        classDiagram
  BaseContainer <|-- Future
  BaseContainer <|-- FutureResult
  FutureBasedN <|-- Future
  FutureResultBasedN <|-- FutureResult
  SupportsKindN <|-- Future
  SupportsKindN <|-- FutureResult
    

async async_identity(instance)

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 (TypeVar(_FirstType))

Return type:

TypeVar(_FirstType)

final class Future(inner_value)

Bases: BaseContainer, SupportsKindN[Future, _ValueType_co, Never, Never], FutureBasedN[_ValueType_co, Never, Never]

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.

See also

• https://gcanti.github.io/fp-ts/modules/Task.ts.html

• https://zio.dev/docs/overview/overview_basic_concurrency

Parameters:

inner_value (Awaitable[TypeVar(_ValueType_co, covariant=True)])

async awaitable()

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[TypeVar(_ValueType_co, covariant=True)]

map(function)

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[[TypeVar(_ValueType_co, covariant=True)], TypeVar(_NewValueType)])

Return type:

Future[TypeVar(_NewValueType)]

apply(container)

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[[TypeVar(_ValueType_co, covariant=True)], TypeVar(_NewValueType)], Any, Any])

Return type:

Future[TypeVar(_NewValueType)]

bind(function)

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[[TypeVar(_ValueType_co, covariant=True)], KindN[Future, TypeVar(_NewValueType), Any, Any]])

Return type:

Future[TypeVar(_NewValueType)]

bind_future(function)

Alias for bind method. Part of the FutureBasedN interface.

Parameters:

function (Callable[[TypeVar(_ValueType_co, covariant=True)], KindN[Future, TypeVar(_NewValueType), Any, Any]])

Return type:

Future[TypeVar(_NewValueType)]

bind_async(function)

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[[TypeVar(_ValueType_co, covariant=True)], Awaitable[KindN[Future, TypeVar(_NewValueType), Any, Any]]])

Return type:

Future[TypeVar(_NewValueType)]

bind_async_future(function)

Alias for bind_async method. Part of the FutureBasedN interface.

Parameters:

function (Callable[[TypeVar(_ValueType_co, covariant=True)], Awaitable[KindN[Future, TypeVar(_NewValueType), Any, Any]]])

Return type:

Future[TypeVar(_NewValueType)]

bind_awaitable(function)

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[[TypeVar(_ValueType_co, covariant=True)], Awaitable[TypeVar(_NewValueType)]])

Return type:

Future[TypeVar(_NewValueType)]

bind_io(function)

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[[TypeVar(_ValueType_co, covariant=True)], IO[TypeVar(_NewValueType)]])

Return type:

Future[TypeVar(_NewValueType)]

classmethod do(expr)

Allows working with unwrapped values of containers in a safe way.

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

>>> async def main() -> bool:
...     return await Future.do(
...         first + second
...         async for first in Future.from_value(2)
...         async for second in Future.from_value(3)
...     ) == IO(5)

>>> assert anyio.run(main) is True

See Do Notation to learn more.

Parameters:

expr (AsyncGenerator[TypeVar(_NewValueType), None])

Return type:

Future[TypeVar(_NewValueType)]

classmethod from_value(inner_value)

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 (TypeVar(_NewValueType))

Return type:

Future[TypeVar(_NewValueType)]

classmethod from_future(inner_value)

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[TypeVar(_NewValueType)])

Return type:

Future[TypeVar(_NewValueType)]

classmethod from_io(inner_value)

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[TypeVar(_NewValueType)])

Return type:

Future[TypeVar(_NewValueType)]

classmethod from_future_result(inner_value)

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[TypeVar(_NewValueType), TypeVar(_NewErrorType)])

Return type:

Future[Result[TypeVar(_NewValueType), TypeVar(_NewErrorType)]]

future(function)

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)

Parameters:

function (Callable[[ParamSpec(_FuncParams, bound= None)], Coroutine[TypeVar(_FirstType), TypeVar(_SecondType), TypeVar(_ValueType_co, covariant=True)]])

Return type:

Callable[[ParamSpec(_FuncParams, bound= None)], Future[TypeVar(_ValueType_co, covariant=True)]]

asyncify(function)

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

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

Parameters:

function (Callable[[ParamSpec(_FuncParams, bound= None)], TypeVar(_ValueType_co, covariant=True)])

Return type:

Callable[[ParamSpec(_FuncParams, bound= None)], Coroutine[Any, Any, TypeVar(_ValueType_co, covariant=True)]]

final class FutureResult(inner_value)

Bases: BaseContainer, SupportsKindN[FutureResult, _ValueType_co, _ErrorType_co, Never], FutureResultBasedN[_ValueType_co, _ErrorType_co, Never]

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.

See also

• https://gcanti.github.io/fp-ts/modules/TaskEither.ts.html

• https://zio.dev/docs/overview/overview_basic_concurrency

Parameters:

inner_value (Awaitable[Result[TypeVar(_ValueType_co, covariant=True), TypeVar(_ErrorType_co, covariant=True)]])

async awaitable()

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[TypeVar(_ValueType_co, covariant=True), TypeVar(_ErrorType_co, covariant=True)]

swap()

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[TypeVar(_ErrorType_co, covariant=True), TypeVar(_ValueType_co, covariant=True)]

map(function)

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[[TypeVar(_ValueType_co, covariant=True)], TypeVar(_NewValueType)])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

apply(container)

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[[TypeVar(_ValueType_co, covariant=True)], TypeVar(_NewValueType)], TypeVar(_ErrorType_co, covariant=True), Any])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

bind(function)

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[[TypeVar(_ValueType_co, covariant=True)], KindN[FutureResult, TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True), Any]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

bind_future_result(function)

Alias for bind method. Part of the FutureResultBasedN interface.

Parameters:

function (Callable[[TypeVar(_ValueType_co, covariant=True)], KindN[FutureResult, TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True), Any]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

bind_async(function)

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[[TypeVar(_ValueType_co, covariant=True)], Awaitable[KindN[FutureResult, TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True), Any]]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

bind_async_future_result(function)

Alias for bind_async method. Part of the FutureResultBasedN interface.

Parameters:

function (Callable[[TypeVar(_ValueType_co, covariant=True)], Awaitable[KindN[FutureResult, TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True), Any]]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

bind_awaitable(function)

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[[TypeVar(_ValueType_co, covariant=True)], Awaitable[TypeVar(_NewValueType)]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

bind_result(function)

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[[TypeVar(_ValueType_co, covariant=True)], Result[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

bind_ioresult(function)

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[[TypeVar(_ValueType_co, covariant=True)], IOResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

bind_io(function)

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[[TypeVar(_ValueType_co, covariant=True)], IO[TypeVar(_NewValueType)]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

bind_future(function)

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[[TypeVar(_ValueType_co, covariant=True)], Future[TypeVar(_NewValueType)]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

bind_async_future(function)

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[[TypeVar(_ValueType_co, covariant=True)], Awaitable[Future[TypeVar(_NewValueType)]]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

alt(function)

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[[TypeVar(_ErrorType_co, covariant=True)], TypeVar(_NewErrorType)])

Return type:

FutureResult[TypeVar(_ValueType_co, covariant=True), TypeVar(_NewErrorType)]

lash(function)

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[[TypeVar(_ErrorType_co, covariant=True)], KindN[FutureResult, TypeVar(_ValueType_co, covariant=True), TypeVar(_NewErrorType), Any]])

Return type:

FutureResult[TypeVar(_ValueType_co, covariant=True), TypeVar(_NewErrorType)]

compose_result(function)

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[TypeVar(_ValueType_co, covariant=True), TypeVar(_ErrorType_co, covariant=True)]], KindN[FutureResult, TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True), Any]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_ErrorType_co, covariant=True)]

classmethod do(expr)

Allows working with unwrapped values of containers in a safe way.

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

>>> async def success() -> bool:
...     return await FutureResult.do(
...         first + second
...         async for first in FutureResult.from_value(2)
...         async for second in FutureResult.from_value(3)
...     ) == IOSuccess(5)

>>> assert anyio.run(success) is True

>>> async def failure() -> bool:
...     return await FutureResult.do(
...         first + second
...         async for first in FutureResult.from_value(2)
...         async for second in FutureResult.from_failure(3)
...     ) == IOFailure(3)

>>> assert anyio.run(failure) is True

See Do Notation to learn more.

Parameters:

expr (AsyncGenerator[TypeVar(_NewValueType), None])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_NewErrorType)]

classmethod from_typecast(inner_value)

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[TypeVar(_NewValueType), TypeVar(_NewErrorType)]])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_NewErrorType)]

classmethod from_future(inner_value)

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[TypeVar(_NewValueType)])

Return type:

FutureResult[TypeVar(_NewValueType), Any]

classmethod from_failed_future(inner_value)

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[TypeVar(_NewErrorType)])

Return type:

FutureResult[Any, TypeVar(_NewErrorType)]

classmethod from_future_result(inner_value)

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[TypeVar(_NewValueType), TypeVar(_NewErrorType)])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_NewErrorType)]

classmethod from_io(inner_value)

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[TypeVar(_NewValueType)])

Return type:

FutureResult[TypeVar(_NewValueType), Any]

classmethod from_failed_io(inner_value)

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[TypeVar(_NewErrorType)])

Return type:

FutureResult[Any, TypeVar(_NewErrorType)]

classmethod from_ioresult(inner_value)

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[TypeVar(_NewValueType), TypeVar(_NewErrorType)])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_NewErrorType)]

classmethod from_result(inner_value)

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[TypeVar(_NewValueType), TypeVar(_NewErrorType)])

Return type:

FutureResult[TypeVar(_NewValueType), TypeVar(_NewErrorType)]

classmethod from_value(inner_value)

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 (TypeVar(_NewValueType))

Return type:

FutureResult[TypeVar(_NewValueType), Any]

classmethod from_failure(inner_value)

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 (TypeVar(_NewErrorType))

Return type:

FutureResult[Any, TypeVar(_NewErrorType)]

FutureSuccess(inner_value)

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 (TypeVar(_NewValueType))

Return type:

FutureResult[TypeVar(_NewValueType), Any]

FutureFailure(inner_value)

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 (TypeVar(_NewErrorType))

Return type:

FutureResult[Any, TypeVar(_NewErrorType)]

FutureResultE

Alias for FutureResult[_ValueType_co, Exception].

alias of FutureResult[_ValueType_co, Exception]

future_safe(exceptions)

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 IOFailure, IOSuccess

>>> @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),
...     IOFailure,
... )

You can also use it with explicit exception types as the first argument:

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

>>> @future_safe(exceptions=(ZeroDivisionError,))
... 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),
...     IOFailure,
... )

In this case, only exceptions that are explicitly listed are going to be caught.

Similar to returns.io.impure_safe() and returns.result.safe() decorators, but works with async functions.

Parameters:

exceptions (Callable[[ParamSpec(_FuncParams, bound= None)], Coroutine[TypeVar(_FirstType), TypeVar(_SecondType), TypeVar(_ValueType_co, covariant=True)]] | tuple[type[TypeVar(_ExceptionType, bound= Exception)], ...])

Return type:

Callable[[ParamSpec(_FuncParams, bound= None)], FutureResult[TypeVar(_ValueType_co, covariant=True), Exception]] | Callable[[Callable[[ParamSpec(_FuncParams, bound= None)], Coroutine[TypeVar(_FirstType), TypeVar(_SecondType), TypeVar(_ValueType_co, covariant=True)]]], Callable[[ParamSpec(_FuncParams, bound= None)], FutureResult[TypeVar(_ValueType_co, covariant=True), TypeVar(_ExceptionType, bound= Exception)]]]