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:
You cannot call async function from a sync one
Any unexpectedly thrown exception can ruin your whole event loop
Ugly composition with lots of await statements
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.
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?
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
We now don’t care about errors at all. In this example any error will cancel the whole pipeline
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.
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.
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
.
This decorator converts async def
into FutureResult
,
which means that it becomes:
Full featured Future
like container
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!
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
.
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]
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
.
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.
instance (~_FirstType) –
~_FirstType
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.
See also
inner_value (Awaitable
[+_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 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)
IO
[+_ValueType]
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)
function (Callable
[[+_ValueType], ~_NewValueType]) –
Future
[~_NewValueType]
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')
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)
Alias for bind method. Part of the FutureBasedN interface.
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')
Alias for bind_async method. Part of the FutureBasedN interface.
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)
function (Callable
[[+_ValueType], Awaitable
[~_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.
>>> 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)
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
inner_value (~_NewValueType) –
Future
[~_NewValueType]
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.
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
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))
inner_value (FutureResult
[~_NewValueType, ~_NewErrorType]) –
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)
function (Callable
[[~_FuncParams], Coroutine
[~_FirstType, ~_SecondType, +_ValueType]]) –
Callable
[[~_FuncParams], Future
[+_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:
>>> 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/
function (Callable
[[~_FuncParams], +_ValueType]) –
Callable
[[~_FuncParams], Coroutine
[Any
, Any
, +_ValueType]]
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.
See also
inner_value (Awaitable
[Result
[+_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 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)
IOResult
[+_ValueType, +_ErrorType]
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)
FutureResult
[+_ErrorType, +_ValueType]
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)
function (Callable
[[+_ValueType], ~_NewValueType]) –
FutureResult
[~_NewValueType, +_ErrorType]
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)
container (KindN
[FutureResult
, Callable
[[+_ValueType], ~_NewValueType], +_ErrorType, Any
]) –
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.
>>> 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)
function (Callable
[[+_ValueType], KindN
[FutureResult
, ~_NewValueType, +_ErrorType, Any
]]) –
FutureResult
[~_NewValueType, +_ErrorType]
Alias for bind method. Part of the FutureResultBasedN interface.
function (Callable
[[+_ValueType], KindN
[FutureResult
, ~_NewValueType, +_ErrorType, Any
]]) –
FutureResult
[~_NewValueType, +_ErrorType]
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)
function (Callable
[[+_ValueType], Awaitable
[KindN
[FutureResult
, ~_NewValueType, +_ErrorType, Any
]]]) –
FutureResult
[~_NewValueType, +_ErrorType]
Alias for bind_async method. Part of the FutureResultBasedN interface.
function (Callable
[[+_ValueType], Awaitable
[KindN
[FutureResult
, ~_NewValueType, +_ErrorType, Any
]]]) –
FutureResult
[~_NewValueType, +_ErrorType]
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)
function (Callable
[[+_ValueType], Awaitable
[~_NewValueType]]) –
FutureResult
[~_NewValueType, +_ErrorType]
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')
function (Callable
[[+_ValueType], Result
[~_NewValueType, +_ErrorType]]) –
FutureResult
[~_NewValueType, +_ErrorType]
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')
function (Callable
[[+_ValueType], IOResult
[~_NewValueType, +_ErrorType]]) –
FutureResult
[~_NewValueType, +_ErrorType]
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)
function (Callable
[[+_ValueType], IO
[~_NewValueType]]) –
FutureResult
[~_NewValueType, +_ErrorType]
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)
function (Callable
[[+_ValueType], Future
[~_NewValueType]]) –
FutureResult
[~_NewValueType, +_ErrorType]
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)
function (Callable
[[+_ValueType], Awaitable
[Future
[~_NewValueType]]]) –
FutureResult
[~_NewValueType, +_ErrorType]
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)
function (Callable
[[+_ErrorType], ~_NewErrorType]) –
FutureResult
[+_ValueType, ~_NewErrorType]
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)
function (Callable
[[+_ErrorType], KindN
[FutureResult
, +_ValueType, ~_NewErrorType, Any
]]) –
FutureResult
[+_ValueType, ~_NewErrorType]
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)
function (Callable
[[Result
[+_ValueType, +_ErrorType]], KindN
[FutureResult
, ~_NewValueType, +_ErrorType, Any
]]) –
FutureResult
[~_NewValueType, +_ErrorType]
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)
inner_value (Future
[Result
[~_NewValueType, ~_NewErrorType]]) –
FutureResult
[~_NewValueType, ~_NewErrorType]
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)
inner_value (Future
[~_NewValueType]) –
FutureResult
[~_NewValueType, Any
]
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)
inner_value (Future
[~_NewErrorType]) –
FutureResult
[Any
, ~_NewErrorType]
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.
inner_value (FutureResult
[~_NewValueType, ~_NewErrorType]) –
FutureResult
[~_NewValueType, ~_NewErrorType]
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)
inner_value (IO
[~_NewValueType]) –
FutureResult
[~_NewValueType, Any
]
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)
inner_value (IO
[~_NewErrorType]) –
FutureResult
[Any
, ~_NewErrorType]
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)
inner_value (IOResult
[~_NewValueType, ~_NewErrorType]) –
FutureResult
[~_NewValueType, ~_NewErrorType]
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)
inner_value (Result
[~_NewValueType, ~_NewErrorType]) –
FutureResult
[~_NewValueType, ~_NewErrorType]
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)
inner_value (~_NewValueType) –
FutureResult
[~_NewValueType, Any
]
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)
inner_value (~_NewErrorType) –
FutureResult
[Any
, ~_NewErrorType]
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,
... )
inner_value (~_NewValueType) –
FutureResult
[~_NewValueType, Any
]
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,
... )
inner_value (~_NewErrorType) –
FutureResult
[Any
, ~_NewErrorType]
Alias for a popular case when Result
has Exception
as error type.
alias of returns.future.FutureResult
[returns.future._ValueType
, Exception
]
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.
function (Callable
[[~_FuncParams], Coroutine
[~_FirstType, ~_SecondType, +_ValueType]]) –
Callable
[[~_FuncParams], FutureResult
[+_ValueType, Exception
]]