Context

Dependency injection is a popular software architechture pattern.

It’s main idea is that you provide Inversion of Control and can pass different things into your logic instead of hardcoding you stuff. And by doing this you are on your way to achieve Single Responsibility for your functions and objects.

Using the context

A lot of programms we write rely on the context implicitly or explicitly. We can rely on confugration, env variables, stubs, logical dependencies, etc.

Let’s look at the example.

Simple app

One of the most popular errors Python developers do in Django is that they overuse settings object inside the business logic. This makes your logic framework-oriented and hard to reason about in large projects.

Because values just pop out of nowhere in a deeply nested functions. And can be changed from the outside, from the context of your app.

Imagine that you have a django based game, where you award users with points for each guessed letter in a word (unguessed letters are marked as '.'):

from django.http import HttpRequest, HttpResponse
from words_app.logic import calculate_points

def view(request: HttpRequest) -> HttpResponse:
    user_word: str = request.POST['word']  # just an example
    points = calculate_points(user_word)
    ...  # later you show the result to user somehow
# Somewhere in your `words_app/logic.py`:

def calculate_points(word: str) -> int:
    guessed_letters_count = len([letter for letter in word if letter != '.'])
    return _award_points_for_letters(guessed_letters_count)

def _award_points_for_letters(guessed: int) -> int:
    return 0 if guessed < 5 else guessed  # minimum 6 points possible!

Straight and simple!

Adding configuration

But, later you decide to make the game more fun: let’s make the minimal accoutable letters threshold configurable for an extra challenge.

You can just do it directly:

def _award_points_for_letters(guessed: int, threshold: int) -> int:
    return 0 if guessed < threshold else guessed

And now your code won’t simply type-check. Because that’s how our caller looks like:

def calculate_points(word: str) -> int:
    guessed_letters_count = len([letter for letter in word if letter != '.'])
    return _award_points_for_letters(guessed_letters_count)

To fix this calculate_points function (and all other upper caller functions) will have to accept threshold: int as a parameter and pass it to _award_points_for_letters.

Imagine that your large project has multiple things to configure in multiple functions. What a mess it would be!

Ok, you can directly use django.settings (or similar) in your _award_points_for_letters function. And ruin your pure logic with framework-specific details. That’s ugly!

Explicitly reling on context

We have learned that this tiny change showed us that it is not so easy to rely on implicit app context.

And instead of passing parameters for all callstack or using dirty framework specific magic you can use RequiresContext container. That was built just for this case.

Let’s see how our code changes:

from django.conf import settings
from django.http import HttpRequest, HttpResponse
from words_app.logic import calculate_points

def view(request: HttpRequest) -> HttpResponse:
    user_word: str = request.POST['word']  # just an example
    points = calculate_points(user_words)(settings)  # passing the dependencies
    ...  # later you show the result to user somehow
# Somewhere in your `words_app/logic.py`:

from typing_extensions import Protocol
from returns.context import RequiresContext

class _Deps(Protocol):  # we rely on abstractions, not direct values or types
    WORD_THRESHOLD: int

def calculate_points(word: str) -> RequiresContext[_Deps, int]:
    guessed_letters_count = len([letter for letter in word if letter != '.'])
    return _award_points_for_letters(guessed_letters_count)

def _award_points_for_letters(guessed: int) -> RequiresContext[_Deps, int]:
    return RequiresContext(
        lambda deps: 0 if guessed < deps.WORD_THRESHOLD else guessed,
    )

And now you can pass your dependencies in a really direct and explicit way.

ask

Let’s try to configure how we mark our unguessed letters (previously unguessed letters were marked as '.'). Let’s say, we want to change this to be _.

How can we do that with our existing function?

def calculate_points(word: str) -> RequiresContext[_Deps, int]:
    guessed_letters_count = len([letter for letter in word if letter != '.'])
    return _award_points_for_letters(guessed_letters_count)

We are already using RequiresContext, but its dependencies are just hidden from us! We have a special helper for this case: returns.context.Context.ask(), which returns us current dependencies.

The only thing we need to is to properly annotate the type for our case: Context[_Deps].ask() Sadly, currently mypy is not able to infer the dependency type out of the context and we need to explicitly provide it.

Let’s see the final result:

from returns.context import Context, RequiresContext

class _Deps(Protocol):  # we rely on abstractions, not direct values or types
    WORD_THRESHOLD: int
    UNGUESSED_CHAR: str

def calculate_points(word: str) -> RequiresContext[_Deps, int]:
    def factory(deps: _Deps) -> RequiresContext[_Deps, int]:
        guessed_letters_count = len([
            letter for letter in word if letter != deps.UNGUESSED_CHAR
        ])
        return _award_points_for_letters(guessed_letters_count)

    return Context[_Deps].ask().bind(factory)

And now we access the current context from any place in our callstack. Isn’t it convenient?

RequiresContext container

The concept behind RequiresContext container is really simple. It is a container around Callable[[EnvType], ReturnType] function.

By its definition it works with pure functions that never fails.

It can be illustrated as a simple nested function:

>>> from typing import Callable
>>> def first(limit: int) -> Callable[[str], bool]:
...     def inner(deps: str) -> bool:
...         return len(deps) > limit
...     return inner
...
>>> assert first(2)('abc')  # first(limit)(deps)
>>> assert not first(5)('abc')  # first(limit)(deps)

That’s basically enough to make dependency injection possible. But how would you compose first function? Let’s say with the following function:

>>> def bool_to_str(arg: bool) -> str:
...     return 'ok' if arg else 'nope'

It would be hard, knowing that it returns another function to be called later when the context is known.

We can wrap it in RequiresContext container to allow better composition!

>>> from returns.context import RequiresContext

>>> def first(limit: int) -> RequiresContext[str, bool]:
...     def inner(deps: str) -> bool:
...         return len(deps) > limit
...     return RequiresContext(inner)  # wrapping function here!

>>> assert first(1).map(bool_to_str)('abc') == 'ok'
>>> assert first(5).map(bool_to_str)('abc') == 'nope'

There’s how execution flows:

graph LR F1["first(1)"] --> F2["RequiresContext[str, bool]"] F2 --> F3 F3["container('abc')"] --> F4["bool"] F4 --> F5 F5["bool_to_str()"] --> F6["str"]

RequiresContext execution flow.

The rule is: the dependencies are injected at the very last moment in time. And then normal logical execution happens.

RequiresContextResult container

This container is a combintaion of RequiresContext[env, Result[a, b]]. Which means that it is a wrapper around pure function that might fail.

We also added a lot of useful methods for this container, so you can work easily with it:

  • bind_result() allows to bind functions that return Result with just one call

  • bind_context() allows to bind functions that return RequiresContext easily

  • There are also several useful constructors from any possible type

Use it when you work with pure context-related functions that might fail.

RequiresContextIOResult container

This container is a combintaion of RequiresContext[env, IOResult[a, b]]. Which means that it is a wrapper around impure function that might fail.

We also added a lot of useful methods for this container, so you can work easily with it:

  • bind_result() allows to bind functions that return Result with just one call

  • bind_io() allows to bind functions that return IO with just one call

  • bind_ioresult() allows to bind functions that return IOResult with just one call

  • bind_context() allows to bind functions that return RequiresContext easily

  • bind_context_result() allows to bind functions that return RequiresContextResult easily

  • There are also several useful constructors from any possible type

Use it when you work with impure context-related functions that might fail. This is basically the main type that is going to be used in most apps.

RequiresContextFutureResult container

This container is a combintaion of RequiresContext[env, FutureResult[a, b]]. Which means that it is a wrapper around impure async function that might fail.

Here’s how it should be used:

from typing import Callable, Sequence

import anyio  # you wound need to `pip install anyio`
import httpx  # you wound need to `pip install httpx`
from typing_extensions import Final, TypedDict

from returns.context import ContextFutureResult, RequiresContextFutureResultE
from returns.functions import tap
from returns.future import FutureResultE, future_safe
from returns.pipeline import managed
from returns.result import safe

_URL: Final = 'https://jsonplaceholder.typicode.com/posts/{0}'
_Post = TypedDict('_Post', {
    'id': int,
    'userId': int,
    'title': str,
    'body': str,
})
_TitleOnly = TypedDict('_TitleOnly', {'title': str})

def _close(client: httpx.AsyncClient, _) -> FutureResultE[None]:
    return future_safe(client.aclose)()

def _fetch_post(
    post_id: int,
) -> RequiresContextFutureResultE[httpx.AsyncClient, _Post]:
    return ContextFutureResult[httpx.AsyncClient].ask().bind_future_result(
        lambda client: future_safe(client.get)(_URL.format(post_id)),
    ).bind_result(
        safe(tap(httpx.Response.raise_for_status)),
    ).map(
        lambda response: response.json(),
    )

def show_titles(
    number_of_posts: int,
) -> RequiresContextFutureResultE[httpx.AsyncClient, Sequence[_TitleOnly]]:
    return RequiresContextFutureResultE.from_iterable([
        # Notice how easily we compose async and sync functions:
        _fetch_post(post_id).map(lambda post: post['title'])
        # TODO: try `for post_id in {2, 1, 0}:` to see how errors work
        for post_id in range(1, number_of_posts + 1)
    ])

if __name__ == '__main__':
    # Let's fetch 3 titles of posts asynchronously:
    managed_httpx = managed(show_titles(3), _close)
    future_result = managed_httpx(
        FutureResultE.from_value(httpx.AsyncClient(timeout=5)),
    )
    print(anyio.run(future_result.awaitable))
    # <IOResult: <Success: (
    #    'sunt aut facere repellat provident occaecati ...',
    #    'qui est esse',
    #    'ea molestias quasi exercitationem repellat qui ipsa sit aut',
    # )>>

This example illustrates the whole point of our actions: writting sync code that executes asynchronously without any magic at all!

We also added a lot of useful methods for this container, so you can work easily with it.

These methods are identical with RequiresContextIOResult:

  • bind_result() allows to bind functions that return Result with just one call

  • bind_io() allows to bind functions that return IO with just one call

  • bind_ioresult() allows to bind functions that return IOResult with just one call

  • bind_future_result() allows to bind functions that return FutureResult with just one call

  • bind_context() allows to bind functions that return RequiresContext easily

  • bind_context_result() allows to bind functions that return RequiresContextResult easily

There are new ones:

Use it when you work with impure context-related functions that might fail. This is basically the main type that is going to be used in most apps.

Aliases

There are several useful alises for RequiresContext and friends with some common values:

  • Reader is an alias for RequiresContext[...] to save you some typing. Uses Reader because it is a native name for this concept from Haskell.

  • RequiresContextResultE is an alias for RequiresContextResult[..., Exception], just use it when you want to work with RequiresContextResult containers that use exceptions as error type. It is named ResultE because it is ResultException and ResultError at the same time.

  • ReaderResult is an alias for RequiresContextResult[...] to save you some typing.

  • ReaderResultE is an alias for RequiresContextResult[..., Exception]

FAQ

How to create unit objects?

RequiresContext requires you to use one of the following methods:

  • from_value when you have a raw value

  • from_requires_context_result when you have RequiresContextResult

  • from_requires_context_ioresult when you have RequiresContextIOResult

RequiresContextResult requires you to use one of the following methods:

  • 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_context when you have successful RequiresContext

  • from_failed_context when you have failed RequiresContext

  • from_typecast when you have RequiresContext[..., Result]

RequiresContextIOResult requires you to use one of the following methods:

  • 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 container

  • from_failed_io when you have failed IO container

  • from_ioresult when you already have IOResult container

  • from_context when you have successful RequiresContext container

  • from_failed_context when you have failed RequiresContext container

  • from_result_context when you have RequiresContextResult container

  • from_typecast when you have RequiresContext[..., IOResult]

RequiresContextFutureResult requires you to use one of the following methods:

  • 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 container

  • from_failed_io when you have failed IO container

  • from_ioresult when you already have IOResult container

  • from_future when you already have successful Future container

  • from_failed_future when you already have failed Future container

  • from_future_result when you already have FutureResult container

  • from_context when you have successful RequiresContext

  • from_failed_context when you have failed RequiresContext

  • from_result_context when you have RequiresContextResult container

  • from_ioresult_context when you have RequiresContextIOResult container

  • from_typecast when you have RequiresContext[..., IOResult]

How can I access dependencies inside the context?

Use .ask() method!

See this guide.

RequiresContext looks like a decorator with arguments

Yes, this container might remind a traditional decorator with arguments, let see an example:

>>> def example(print_result: bool):
...     def decorator(function):
...         def factory(*args, **kwargs):
...             original = function(*args, **kwargs)
...             if print_result:
...                 print(original)
...             return original
...         return factory
...     return decorator

And it can be used like so:

>>> @example(print_result=True)
... def my_function(first: int, second: int) -> int:
...     return first + second

>>> assert my_function(2, 3) == 5
5

We can model the similar idea with RequiresContext:

>>> from returns.context import RequiresContext

>>> def my_function(first: int, second: int) -> RequiresContext[bool, int]:
...     def factory(print_result: bool) -> int:
...         original = first + second
...         if print_result:
...             print(original)
...         return original
...     return RequiresContext(factory)

>>> assert my_function(2, 3)(False) == 5
>>> assert my_function(2, 3)(True) == 5
5

As you can see, it is easier to change the behaviour of a function with RequiresContext. While decorator with arguments glues values to a function forever. Decide when you need which behaviour carefully.

Why can’t we use RequiresContext[e, Result] instead of RequiresContextResult?

We actually can! But, it is harder to write. And RequiresContextResult is actually the very same thing as RequiresContext[e, Result], but has nicer API:

x: RequiresContext[int, Result[int, str]]
x.map(lambda result: result.map(lambda number: number + 1))

# Is the same as:

y: RequiresContextResult[int, int, str]
y.map(lambda number: number + 1)

The second one looks better, doesn’t it? The same applies for RequiresContextIOResult and RequiresContextFutureResult as well.

Why do I have to use explicit type annotation for ask method?

Because mypy cannot possibly know the type of current context. This is hard even for a plugin.

So, using this technique is better:

from returns.context import Context, RequiresContext

def some_context(*args, **kwargs) -> RequiresContext[int, str]:
    def factory(deps: int) -> RequiresContext[int, str]:
        ...
    return Context[int].ask().bind(factory)

What is the difference between DI and RequiresContext?

Dependency Injection pattern and Inversion of Control principle forms a lot of ideas and tooling that do pretty much the same as RequiresContext container.

What is the difference? Why do we need each of them?

Let’s find out! Tools like dependencies or punq tries to:

  1. Inspect (by name or type respectively) function or class that needs dependencies

  2. Build the required dependency tree from the source defined in the service container

There are other tools like inject that also invades your code with @inject decorator.

RequiresContext works completely different. It respects your code and does not try to inspect in any manner. It also does not care about building dependencies at all.

All it does is: provides simple API to compose functions that need additional context (or dependencies) to run.

You can even use them together: RequiresContext will pass depedencies built by dry-python/dependencies (or any other tool of your choice) as a deps parameter to RequiresContext instance.

When to use which? Let’s dig into it!

  • RequiresContext offers explicit context passing for the whole function stack inside your program. This means two things: you will have to pass it through all your code and use it everywhere inside your program explicitly, when you need to access the environment and dependencies

  • Traditional DI allows to leave a lot of code unaware of dependency injection. Because you don’t have to maintain the context everywhere. You just need to adjust your API to meet the dependency injector requirements. On the other hand, you lose explicitness here.

So when to use RequiresContext?

  1. When you write pure functional code

  2. When you want to know which code relies on context and which is free from it, RequiresContext makes this explicit and typed

  3. When you rely on types inside your program

  4. When you want to rely on functions rather than magic

When not to use RequiresContext and use traditional DI?

  1. When you already have a lot of code written in a different approach: in OOP and/or imperative styles

  2. When you need to pass dependencies into a very deep level of your call stack implicitly (without modifing the whole stack), this is called magic

  3. When you not rely on types for dependencies. There are cases when DI is made by names or tags

Here’s an example that might give you a better understanding of how RequiresContext is used on real and rather big projects:

from typing import Callable, Dict, Protocol, final
from returns.io import IOResultE
from returns.context import ReaderIOResultE

@final
class _SyncPermissionsDeps(Protocol):
    fetch_metadata: Callable[[], IOResultE['Metadata']]
    get_user_permissions: Callable[['Metadata'], Dict[int, str]]  # pure
    update_bi_permissions: Callable[[Dict[int, str]], IOResultE['Payload']]

def sync_permissions() -> ReaderIOResultE[_SyncPermissionsDeps, 'Payload']:
    """
    This functions runs a scheduled task once a day.

    It syncs permissions from the metadata storage to our BI system.
    """
    def factory(deps: _SyncPermissionsDeps) -> IOResultE['Payload']:
        return deps.fetch_metadata().map(
            deps.get_user_permissions,
        ).bind_ioresult(
            deps.update_bi_permissions,
        )
    return ReaderIOResult(factory)

And then it is called like so:

# tasks.py
from celery import shared_task
from returns.functions import raise_exception

from logic.usecases.sync_permissions import sync_permissions
from infrastructure.implemented import Container
from infrastructure.services import bi
from infrastructure.repositories import db

@shared_task(autoretry_for=(ConnectionError,), max_retries=3)
def queue_sync_permissions():
    # Building the container with dependencies to pass it into the context.
    # We also make sure that we don't forget to raise internal exceptions
    # and trigger celery retries.
    return sync_permissions().fix(raise_exception)(Container(
        fetch_metadata=db.select_user_metadata,
        get_user_permissions=bi.permissions_from_user,
        update_bi_permissions=bi.put_user_permissions,
    ))

API Reference

RequiresContext

graph TD; RequiresContext Context BaseContainer --> RequiresContext Generic --> RequiresContext Immutable --> Context Generic --> Context
NoDeps = typing.Any

Sometimes RequiresContext and other similar types might be used with no explicit dependencies so we need to have this type alias for Any.

class RequiresContext(inner_value)[source]

Bases: returns.primitives.container.BaseContainer, typing.Generic

The RequiresContext container.

It’s main purpose is to wrap some specific function and to provide tools to compose other functions around it without actually calling it.

The RequiresContext container passes the state you want to share between functions. Functions may read that state, but can’t change it. The RequiresContext container lets us access shared immutable state within a specific context.

It can be used for lazy evaluation and typed dependency injection.

RequiresContext is used with functions that never fail. If you want to use RequiresContext with returns Result then consider using RequiresContextResult instead.

Note

This container does not wrap ANY value. It wraps only functions. You won’t be able to supply arbitrary types!

Parameters

inner_value (Callable[[-_EnvType], +_ReturnType]) –

empty: ClassVar[Any] = <object object>

A convinient placeholder to call methods created by .from_value():

map(function)[source]

Allows to compose functions inside the wrapped container.

Here’s how it works:

>>> from returns.context import RequiresContext
>>> def first(lg: bool) -> RequiresContext[float, int]:
...     # `deps` has `float` type here:
...     return RequiresContext(
...         lambda deps: deps if lg else -deps,
...     )

>>> assert first(True).map(lambda number: number * 10)(2.5) == 25.0
>>> assert first(False).map(lambda number: number * 10)(0.1) -1.0
Parameters

function (Callable[[+_ReturnType], ~_NewReturnType]) –

Return type

RequiresContext[-_EnvType, ~_NewReturnType]

apply(container)[source]

Calls a wrapped function in a container on this container.

>>> from returns.context import RequiresContext
>>> assert RequiresContext.from_value('a').apply(
...    RequiresContext.from_value(lambda inner: inner + 'b')
... )(...) == 'ab'
Parameters

container (RequiresContext[-_EnvType, Callable[[+_ReturnType], ~_NewReturnType]]) –

Return type

RequiresContext[-_EnvType, ~_NewReturnType]

bind(function)[source]

Composes a container with a function returning another container.

This is useful when you do several computations that rely on the same context.

>>> from returns.context import RequiresContext

>>> def first(lg: bool) -> RequiresContext[float, int]:
...     # `deps` has `float` type here:
...     return RequiresContext(
...         lambda deps: deps if lg else -deps,
...     )

>>> def second(number: int) -> RequiresContext[float, str]:
...     # `deps` has `float` type here:
...     return RequiresContext(
...         lambda deps: '>=' if number >= deps else '<',
...     )

>>> assert first(True).bind(second)(1) == '>='
>>> assert first(False).bind(second)(2) == '<'
Parameters

function (Callable[[+_ReturnType], RequiresContext[-_EnvType, ~_NewReturnType]]) –

Return type

RequiresContext[-_EnvType, ~_NewReturnType]

classmethod from_value(inner_value)[source]

Used to return some specific value from the container.

Consider this method as some kind of factory. Passed value will be a return type. Make sure to use empty for getting the unit value.

>>> from returns.context import RequiresContext
>>> unit = RequiresContext.from_value(5)
>>> assert unit(RequiresContext.empty) == 5

Might be used with or without direct type hint.

Part of the returns.primitives.interfaces.Applicative protocol.

Parameters

inner_value (~_FirstType) –

Return type

RequiresContext[Any, ~_FirstType]

classmethod from_iterable(inner_value)[source]

Transforms an iterable of RequiresContext containers.

Returns a single container with a sequence of values.

>>> from returns.context import RequiresContext

>>> assert RequiresContext.from_iterable([
...    RequiresContext.from_value(1),
...    RequiresContext.from_value(2),
... ])(...) == (1, 2)
Parameters

inner_value (Iterable[RequiresContext[-_EnvType, ~_ValueType]]) –

Return type

RequiresContext[-_EnvType, Sequence[~_ValueType]]

classmethod from_requires_context_result(inner_value)[source]

Typecasts RequiresContextResult to RequiresContext instance.

Breaks RequiresContextResult[e, a, b] into RequiresContext[e, Result[a, b]].

>>> from returns.context import RequiresContext
>>> from returns.context import RequiresContextResult
>>> from returns.result import Success
>>> assert RequiresContext.from_requires_context_result(
...    RequiresContextResult.from_value(1),
... )(...) == Success(1)

Can be reverted with RequiresContextResult.from_typecast.

Parameters

inner_value (RequiresContextResult[_EnvType, _ValueType, _ErrorType]) –

Return type

RequiresContext[_EnvType, Result[_ValueType, _ErrorType]]

classmethod from_requires_context_ioresult(inner_value)[source]

Typecasts RequiresContextIOResult to RequiresContext instance.

Breaks RequiresContextIOResult[e, a, b] into RequiresContext[e, IOResult[a, b]].

>>> from returns.context import RequiresContext
>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess
>>> assert RequiresContext.from_requires_context_ioresult(
...    RequiresContextIOResult.from_value(1),
... )(...) == IOSuccess(1)

Can be reverted with RequiresContextIOResult.from_typecast.

Parameters

inner_value (RequiresContextIOResult[_EnvType, _ValueType, _ErrorType]) –

Return type

RequiresContext[_EnvType, IOResult[_ValueType, _ErrorType]]

class Context(*args, **kwds)[source]

Bases: returns.primitives.types.Immutable, typing.Generic

Helpers that can be used to work with RequiresContext container.

Some of them require an explicit type to be specified.

This class contains methods that require to explicitly set type annotations. Why? Because it is impossible to figure out the type without them.

So, here’s how you should use them:

Context[Dict[str, str]].ask()

Otherwise, your .ask() method will return RequiresContext[<nothing>, <nothing>], which is unusable:

env = Context.ask()
env(some_deps)

And mypy will warn you: error: Need type annotation for 'a'

classmethod ask()[source]

Get current context to use the dependencies.

It is a common scenario when you need to use the environment. For example, you want to do some context-related computation, but you don’t have the context instance at your disposal. That’s where .ask() becomes useful!

>>> from typing_extensions import TypedDict
>>> class Deps(TypedDict):
...     message: str

>>> def first(lg: bool) -> RequiresContext[Deps, int]:
...     # `deps` has `Deps` type here:
...     return RequiresContext(
...         lambda deps: deps['message'] if lg else 'error',
...     )

>>> def second(text: str) -> RequiresContext[int, int]:
...     return first(len(text) > 3)

>>> assert second('abc')({'message': 'ok'}) == 'error'
>>> assert second('abcd')({'message': 'ok'}) == 'ok'

And now imagine that you have to change this 3 limit. And you want to be able to set it via environment as well. Ok, let’s fix it with the power of Context.ask()!

>>> from typing_extensions import TypedDict
>>> class Deps(TypedDict):
...     message: str
...     limit: int   # note this new field!

>>> def new_first(lg: bool) -> RequiresContext[Deps, int]:
...     # `deps` has `Deps` type here:
...     return RequiresContext(
...         lambda deps: deps['message'] if lg else 'err',
...     )

>>> def new_second(text: str) -> RequiresContext[int, int]:
...     return Context[Deps].ask().bind(
...         lambda deps: new_first(len(text) > deps.get('limit', 3)),
...     )

>>> assert new_second('abc')({'message': 'ok', 'limit': 2}) == 'ok'
>>> assert new_second('abcd')({'message': 'ok'}) == 'ok'
>>> assert new_second('abcd')({'message': 'ok', 'limit': 5}) == 'err'

That’s how ask works.

Return type

RequiresContext[-_EnvType, -_EnvType]

Reader

Sometimes RequiresContext is too long to type.

Parameters

inner_value (Callable[[-_EnvType], +_ReturnType]) –

alias of returns.context.requires_context.RequiresContext

RequiresContextResult

graph TD; RequiresContextResult ContextResult BaseContainer --> RequiresContextResult Generic --> RequiresContextResult Immutable --> ContextResult Generic --> ContextResult
class RequiresContextResult(inner_value)[source]

Bases: returns.primitives.container.BaseContainer, typing.Generic

The RequiresContextResult combinator.

See returns.context.requires_context.RequiresContext for more docs.

This is just a handy wrapper around RequiresContext[env, Result[a, b]] which represents a context-dependent pure operation that might fail and return returns.result.Result.

It has several important differences from the regular Result classes. It does not have Success and Failure subclasses. Because, the computation is not yet performed. And we cannot know the type in advance.

So, this is a thin wrapper, without any changes in logic.

Why do we need this wrapper? That’s just for better usability!

>>> from returns.context import RequiresContext
>>> from returns.result import Success, Result

>>> def function(arg: int) -> Result[int, str]:
...      return Success(arg + 1)

>>> # Without wrapper:
>>> assert RequiresContext.from_value(Success(1)).map(
...     lambda result: result.bind(function),
... )(...) == Success(2)

>>> # With wrapper:
>>> assert RequiresContextResult.from_value(1).bind_result(
...     function,
... )(...) == Success(2)

This way RequiresContextResult allows to simply work with:

  • raw values and pure functions

  • RequiresContext values and pure functions returning it

  • Result and functions returning it

Important implementation detail: due it is meaning, RequiresContextResult cannot have Success and Failure subclasses.

We only have just one type. That’s by design.

Different converters are also not supported for this type. Use converters inside the RequiresContext context, not outside.

Parameters

inner_value (Callable[[-_EnvType], Result[+_ValueType, +_ErrorType]]) –

empty: ClassVar[Any] = <object object>

A convinient placeholder to call methods created by .from_value().

map(function)[source]

Composes successful container with a pure function.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success, Failure

>>> assert RequiresContextResult.from_value(1).map(
...     lambda x: x + 1,
... )(...) == Success(2)

>>> assert RequiresContextResult.from_failure(1).map(
...     lambda x: x + 1,
... )(...) == Failure(1)
Parameters

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

Return type

RequiresContextResult[-_EnvType, ~_NewValueType, +_ErrorType]

apply(container)[source]

Calls a wrapped function in a container on this container.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success, Failure, Result

>>> def transform(arg: str) -> str:
...     return arg + 'b'

>>> assert RequiresContextResult.from_value('a').apply(
...    RequiresContextResult.from_value(transform),
... )(...) == Success('ab')

>>> assert RequiresContextResult.from_failure('a').apply(
...    RequiresContextResult.from_value(transform),
... )(...) == Failure('a')

>>> assert isinstance(RequiresContextResult.from_value('a').apply(
...    RequiresContextResult.from_failure(transform),
... )(...), Result.failure_type) is True
Parameters

container (RequiresContextResult[-_EnvType, Callable[[+_ValueType], ~_NewValueType], +_ErrorType]) –

Return type

RequiresContextResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind(function)[source]

Composes this container with a function returning the same type.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success, Failure

>>> def first(lg: bool) -> RequiresContextResult[float, int, int]:
...     # `deps` has `float` type here:
...     return RequiresContextResult(
...         lambda deps: Success(deps) if lg else Failure(-deps),
...     )

>>> def second(
...     number: int,
... ) -> RequiresContextResult[float, str, int]:
...     # `deps` has `float` type here:
...     return RequiresContextResult(
...         lambda deps: Success('>=' if number >= deps else '<'),
...     )

>>> assert first(True).bind(second)(1) == Success('>=')
>>> assert first(False).bind(second)(2) == Failure(-2)
Parameters

function (Callable[[+_ValueType], RequiresContextResult[-_EnvType, ~_NewValueType, +_ErrorType]]) –

Return type

RequiresContextResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_result(function)[source]

Binds Result returning function to current container.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success, Failure, Result

>>> def function(num: int) -> Result[int, str]:
...     return Success(num + 1) if num > 0 else Failure('<0')

>>> assert RequiresContextResult.from_value(1).bind_result(
...     function,
... )(RequiresContextResult.empty) == Success(2)

>>> assert RequiresContextResult.from_value(0).bind_result(
...     function,
... )(RequiresContextResult.empty) == Failure('<0')

>>> assert RequiresContextResult.from_failure(':(').bind_result(
...     function,
... )(RequiresContextResult.empty) == Failure(':(')
Parameters

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

Return type

RequiresContextResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_context(function)[source]

Binds RequiresContext returning function to current container.

>>> from returns.context import RequiresContext
>>> from returns.result import Success, Failure

>>> def function(arg: int) -> RequiresContext[str, int]:
...     return RequiresContext(lambda deps: len(deps) + arg)

>>> assert function(2)('abc') == 5

>>> assert RequiresContextResult.from_value(2).bind_context(
...     function,
... )('abc') == Success(5)

>>> assert RequiresContextResult.from_failure(2).bind_context(
...     function,
... )('abc') == Failure(2)
Parameters

function (Callable[[+_ValueType], ForwardRef]) –

Return type

RequiresContextResult[_EnvType, _NewValueType, _ErrorType]

fix(function)[source]

Composes failed container with a pure function.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success

>>> assert RequiresContextResult.from_value(1).fix(
...     lambda x: x + 1,
... )(...) == Success(1)

>>> assert RequiresContextResult.from_failure(1).fix(
...     lambda x: x + 1,
... )(...) == Success(2)
Parameters

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

Return type

RequiresContextResult[-_EnvType, ~_NewValueType, +_ErrorType]

alt(function)[source]

Composes failed container with a pure function.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success, Failure

>>> assert RequiresContextResult.from_value(1).alt(
...     lambda x: x + 1,
... )(...) == Success(1)

>>> assert RequiresContextResult.from_failure(1).alt(
...     lambda x: x + 1,
... )(...) == Failure(2)
Parameters

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

Return type

RequiresContextResult[-_EnvType, +_ValueType, ~_NewErrorType]

rescue(function)[source]

Composes this container with a function returning the same type.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success, Failure

>>> def rescuable(arg: str) -> RequiresContextResult[str, str, str]:
...      if len(arg) > 1:
...          return RequiresContextResult(
...              lambda deps: Success(deps + arg),
...          )
...      return RequiresContextResult(
...          lambda deps: Failure(arg + deps),
...      )

>>> assert RequiresContextResult.from_value('a').rescue(
...     rescuable,
... )('c') == Success('a')
>>> assert RequiresContextResult.from_failure('a').rescue(
...     rescuable,
... )('c') == Failure('ac')
>>> assert RequiresContextResult.from_failure('aa').rescue(
...     rescuable,
... )('b') == Success('baa')
Parameters

function (Callable[[+_ErrorType], RequiresContextResult[-_EnvType, +_ValueType, ~_NewErrorType]]) –

Return type

RequiresContextResult[-_EnvType, +_ValueType, ~_NewErrorType]

value_or(default_value)[source]

Returns a callable that either returns a success or default value.

>>> from returns.context import RequiresContextResult

>>> assert RequiresContextResult.from_value(1).value_or(2)(
...     RequiresContextResult.empty,
... ) == 1

>>> assert RequiresContextResult.from_failure(1).value_or(2)(
...     RequiresContextResult.empty,
... ) == 2
Parameters

default_value (~_FirstType) –

Return type

Callable[[-_EnvType], Union[+_ValueType, ~_FirstType]]

unwrap()[source]

Returns a callable that unwraps success value or raises exception.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success, Failure

>>> assert RequiresContextResult(
...    lambda _: Success(1),
... ).unwrap()(RequiresContextResult.empty) == 1

>>> RequiresContextResult(
...    lambda _: Failure(1),
... ).unwrap()(RequiresContextResult.empty)
Traceback (most recent call last):
  ...
returns.primitives.exceptions.UnwrapFailedError
Return type

Callable[[-_EnvType], +_ValueType]

failure()[source]

Returns a callable that unwraps failure value or raises exception.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success, Failure

>>> assert RequiresContextResult(
...    lambda _: Failure(1),
... ).failure()(RequiresContextResult.empty) == 1

>>> RequiresContextResult(
...    lambda _: Success(1),
... ).failure()(RequiresContextResult.empty)
Traceback (most recent call last):
  ...
returns.primitives.exceptions.UnwrapFailedError
Return type

Callable[[-_EnvType], +_ErrorType]

classmethod from_result(inner_value)[source]

Creates new container with Result as a unit value.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success, Failure
>>> deps = RequiresContextResult.empty

>>> assert RequiresContextResult.from_result(
...    Success(1),
... )(deps) == Success(1)

>>> assert RequiresContextResult.from_result(
...    Failure(1),
... )(deps) == Failure(1)
Parameters

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

Return type

RequiresContextResult[Any, +_ValueType, +_ErrorType]

classmethod from_typecast(inner_value)[source]

You might end up with RequiresContext[Result[...]] as a value.

This method is designed to turn it into RequiresContextResult. It will save all the typing information.

It is just more useful!

>>> from returns.context import RequiresContext
>>> from returns.result import Success, Failure

>>> assert RequiresContextResult.from_typecast(
...     RequiresContext.from_value(Success(1)),
... )(RequiresContextResult.empty) == Success(1)

>>> assert RequiresContextResult.from_typecast(
...     RequiresContext.from_value(Failure(1)),
... )(RequiresContextResult.empty) == Failure(1)
Parameters

inner_value (RequiresContext[_EnvType, Result[_NewValueType, _NewErrorType]]) –

Return type

RequiresContextResult[_EnvType, _NewValueType, _NewErrorType]

classmethod from_context(inner_value)[source]

Creates new container from RequiresContext as a success unit.

>>> from returns.context import RequiresContext
>>> from returns.result import Success
>>> assert RequiresContextResult.from_context(
...     RequiresContext.from_value(1),
... )(...) == Success(1)
Parameters

inner_value (RequiresContext[_EnvType, _FirstType]) –

Return type

RequiresContextResult[_EnvType, _FirstType, Any]

classmethod from_failed_context(inner_value)[source]

Creates new container from RequiresContext as a failure unit.

>>> from returns.context import RequiresContext
>>> from returns.result import Failure
>>> assert RequiresContextResult.from_failed_context(
...     RequiresContext.from_value(1),
... )(...) == Failure(1)
Parameters

inner_value (RequiresContext[_EnvType, _FirstType]) –

Return type

RequiresContextResult[_EnvType, Any, _FirstType]

classmethod from_value(inner_value)[source]

Creates new container with Success(inner_value) as a unit value.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success
>>> assert RequiresContextResult.from_value(1)(...) == Success(1)
Parameters

inner_value (~_FirstType) –

Return type

RequiresContextResult[Any, ~_FirstType, Any]

classmethod from_failure(inner_value)[source]

Creates new container with Failure(inner_value) as a unit value.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Failure
>>> assert RequiresContextResult.from_failure(1)(...) == Failure(1)
Parameters

inner_value (~_FirstType) –

Return type

RequiresContextResult[Any, Any, ~_FirstType]

classmethod from_iterable(inner_value)[source]

Transforms an iterable of RequiresContextResult containers.

Returns a single container with multiple elements inside.

>>> from returns.context import RequiresContextResult
>>> from returns.result import Success, Failure

>>> assert RequiresContextResult.from_iterable([
...    RequiresContextResult.from_value(1),
...    RequiresContextResult.from_value(2),
... ])(...) == Success((1, 2))

>>> assert RequiresContextResult.from_iterable([
...    RequiresContextResult.from_value(1),
...    RequiresContextResult.from_failure('a'),
... ])(...) == Failure('a')

>>> assert RequiresContextResult.from_iterable([
...    RequiresContextResult.from_failure('a'),
...    RequiresContextResult.from_value(1),
... ])(...) == Failure('a')
Parameters

inner_value (Iterable[RequiresContextResult[-_EnvType, +_ValueType, +_ErrorType]]) –

Return type

RequiresContextResult[-_EnvType, Sequence[+_ValueType], +_ErrorType]

class ContextResult(*args, **kwds)[source]

Bases: returns.primitives.types.Immutable, typing.Generic

Helpers that can be used to work with RequiresContextResult container.

Related to returns.context.Context, refer there for the docs.

classmethod ask()[source]

Is used to get the current dependencies inside the call stack.

Similar to returns.context.Context.ask(), but returns Result instead of a regular value.

Please, refer to the docs there to learn how to use it.

One important note that is worth duplicating here: you might need to provide _EnvType explicitly, so mypy will know about it statically.

>>> from returns.context import ContextResult
>>> from returns.result import Success
>>> assert ContextResult[int].ask().map(str)(1) == Success('1')
Return type

RequiresContextResult[-_EnvType, -_EnvType, Any]

RequiresContextResultE

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

alias of returns.context.requires_context_result.RequiresContextResult[_EnvType, _ValueType, Exception]

ReaderResult

Alias to save you some typing. Uses original name from Haskell.

alias of returns.context.requires_context_result.RequiresContextResult[_EnvType, _ValueType, _ErrorType]

ReaderResultE

Alias to save you some typing. Has Exception as error type.

alias of returns.context.requires_context_result.RequiresContextResult[_EnvType, _ValueType, Exception]

RequiresContextIOResult

graph TD; ContextIOResult RequiresContextIOResult BaseContainer --> RequiresContextIOResult Generic --> RequiresContextIOResult Immutable --> ContextIOResult Generic --> ContextIOResult
class RequiresContextIOResult(inner_value)[source]

Bases: returns.primitives.container.BaseContainer, typing.Generic

The RequiresContextIOResult combinator.

See returns.context.requires_context.RequiresContext and returns.context.requires_context_result.RequiresContextResult for more docs.

This is just a handy wrapper around RequiresContext[env, IOResult[a, b]] which represents a context-dependent impure operation that might fail.

It has several important differences from the regular Result classes. It does not have Success and Failure subclasses. Because, the computation is not yet performed. And we cannot know the type in advance.

So, this is a thin wrapper, without any changes in logic.

Why do we need this wrapper? That’s just for better usability!

>>> from returns.context import RequiresContext
>>> from returns.io import IOSuccess, IOResult

>>> def function(arg: int) -> IOResult[int, str]:
...      return IOSuccess(arg + 1)

>>> # Without wrapper:
>>> assert RequiresContext.from_value(IOSuccess(1)).map(
...     lambda ioresult: ioresult.bind(function),
... )(...) == IOSuccess(2)

>>> # With wrapper:
>>> assert RequiresContextIOResult.from_value(1).bind_ioresult(
...     function,
... )(...) == IOSuccess(2)

This way RequiresContextIOResult allows to simply work with:

  • raw values and pure functions

  • RequiresContext values and pure functions returning it

  • RequiresContextResult values and pure functions returning it

  • Result and pure functions returning it

  • IOResult and functions returning it

  • other RequiresContextIOResult related functions and values

This is a complex type for complex tasks!

Important implementation detail: due it is meaning, RequiresContextIOResult cannot have Success and Failure subclasses.

We only have just one type. That’s by design.

Different converters are also not supported for this type. Use converters inside the RequiresContext context, not outside.

Parameters

inner_value (Callable[[-_EnvType], IOResult[+_ValueType, +_ErrorType]]) –

empty: ClassVar[Any] = <object object>

A convinient placeholder to call methods created by .from_value().

map(function)[source]

Composes successful container with a pure function.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess, IOFailure

>>> assert RequiresContextIOResult.from_value(1).map(
...     lambda x: x + 1,
... )(...) == IOSuccess(2)

>>> assert RequiresContextIOResult.from_failure(1).map(
...     lambda x: x + 1,
... )(...) == IOFailure(1)
Parameters

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

Return type

RequiresContextIOResult[-_EnvType, ~_NewValueType, +_ErrorType]

apply(container)[source]

Calls a wrapped function in a container on this container.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess, IOFailure, IOResult

>>> def transform(arg: str) -> str:
...     return arg + 'b'

>>> assert RequiresContextIOResult.from_value('a').apply(
...    RequiresContextIOResult.from_value(transform),
... )(...) == IOSuccess('ab')

>>> assert RequiresContextIOResult.from_failure('a').apply(
...    RequiresContextIOResult.from_value(transform),
... )(...) == IOFailure('a')

>>> assert isinstance(RequiresContextIOResult.from_value('a').apply(
...    RequiresContextIOResult.from_failure(transform),
... )(...), IOResult.failure_type) is True
Parameters

container (RequiresContextIOResult[-_EnvType, Callable[[+_ValueType], ~_NewValueType], +_ErrorType]) –

Return type

RequiresContextIOResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind(function)[source]

Composes this container with a function returning the same type.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess, IOFailure

>>> def first(lg: bool) -> RequiresContextIOResult[float, int, int]:
...     # `deps` has `float` type here:
...     return RequiresContextIOResult(
...         lambda deps: IOSuccess(deps) if lg else IOFailure(-deps),
...     )

>>> def second(
...     number: int,
... ) -> RequiresContextIOResult[float, str, int]:
...     # `deps` has `float` type here:
...     return RequiresContextIOResult(
...         lambda deps: IOSuccess('>=' if number >= deps else '<'),
...     )

>>> assert first(True).bind(second)(1) == IOSuccess('>=')
>>> assert first(False).bind(second)(2) == IOFailure(-2)
Parameters

function (Callable[[+_ValueType], RequiresContextIOResult[-_EnvType, ~_NewValueType, +_ErrorType]]) –

Return type

RequiresContextIOResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_result(function)[source]

Binds Result returning function to the current container.

>>> from returns.context import RequiresContextIOResult
>>> from returns.result import Success, Failure, Result
>>> from returns.io import IOSuccess, IOFailure

>>> def function(num: int) -> Result[int, str]:
...     return Success(num + 1) if num > 0 else Failure('<0')

>>> assert RequiresContextIOResult.from_value(1).bind_result(
...     function,
... )(RequiresContextIOResult.empty) == IOSuccess(2)

>>> assert RequiresContextIOResult.from_value(0).bind_result(
...     function,
... )(RequiresContextIOResult.empty) == IOFailure('<0')

>>> assert RequiresContextIOResult.from_failure(':(').bind_result(
...     function,
... )(RequiresContextIOResult.empty) == IOFailure(':(')
Parameters

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

Return type

RequiresContextIOResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_context(function)[source]

Binds RequiresContext returning function to current container.

>>> from returns.context import RequiresContext
>>> from returns.io import IOSuccess, IOFailure

>>> def function(arg: int) -> RequiresContext[str, int]:
...     return RequiresContext(lambda deps: len(deps) + arg)

>>> assert function(2)('abc') == 5

>>> assert RequiresContextIOResult.from_value(2).bind_context(
...     function,
... )('abc') == IOSuccess(5)

>>> assert RequiresContextIOResult.from_failure(2).bind_context(
...     function,
... )('abc') == IOFailure(2)
Parameters

function (Callable[[+_ValueType], ForwardRef]) –

Return type

RequiresContextIOResult[_EnvType, _NewValueType, _ErrorType]

bind_context_result(function)[source]

Binds RequiresContextResult returning function to the current one.

>>> from returns.context import RequiresContextResult
>>> from returns.io import IOSuccess, IOFailure
>>> from returns.result import Success, Failure

>>> def function(arg: int) -> RequiresContextResult[str, int, int]:
...     if arg > 0:
...         return RequiresContextResult(
...             lambda deps: Success(len(deps) + arg),
...         )
...     return RequiresContextResult(
...         lambda deps: Failure(len(deps) + arg),
...     )

>>> assert function(2)('abc') == Success(5)
>>> assert function(-1)('abc') == Failure(2)

>>> assert RequiresContextIOResult.from_value(
...    2,
... ).bind_context_result(
...     function,
... )('abc') == IOSuccess(5)

>>> assert RequiresContextIOResult.from_value(
...    -1,
... ).bind_context_result(
...     function,
... )('abc') == IOFailure(2)

>>> assert RequiresContextIOResult.from_failure(
...    2,
... ).bind_context_result(
...     function,
... )('abc') == IOFailure(2)
Parameters

function (Callable[[+_ValueType], ForwardRef]) –

Return type

RequiresContextIOResult[_EnvType, _NewValueType, _ErrorType]

bind_io(function)[source]

Binds IO returning function to the current container.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IO, IOSuccess, IOFailure

>>> def function(number: int) -> IO[str]:
...     return IO(str(number))

>>> assert RequiresContextIOResult.from_value(1).bind_io(
...     function,
... )(RequiresContextIOResult.empty) == IOSuccess('1')

>>> assert RequiresContextIOResult.from_failure(1).bind_io(
...     function,
... )(RequiresContextIOResult.empty) == IOFailure(1)
Parameters

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

Return type

RequiresContextIOResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_ioresult(function)[source]

Binds IOResult returning function to the current container.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOResult, IOSuccess, IOFailure

>>> def function(num: int) -> IOResult[int, str]:
...     return IOSuccess(num + 1) if num > 0 else IOFailure('<0')

>>> assert RequiresContextIOResult.from_value(1).bind_ioresult(
...     function,
... )(RequiresContextIOResult.empty) == IOSuccess(2)

>>> assert RequiresContextIOResult.from_value(0).bind_ioresult(
...     function,
... )(RequiresContextIOResult.empty) == IOFailure('<0')

>>> assert RequiresContextIOResult.from_failure(':(').bind_ioresult(
...     function,
... )(RequiresContextIOResult.empty) == IOFailure(':(')
Parameters

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

Return type

RequiresContextIOResult[-_EnvType, ~_NewValueType, +_ErrorType]

fix(function)[source]

Composes failed container with a pure function.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess

>>> assert RequiresContextIOResult.from_value(1).fix(
...     lambda x: x + 1,
... )(...) == IOSuccess(1)

>>> assert RequiresContextIOResult.from_failure(1).fix(
...     lambda x: x + 1,
... )(...) == IOSuccess(2)
Parameters

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

Return type

RequiresContextIOResult[-_EnvType, ~_NewValueType, +_ErrorType]

alt(function)[source]

Composes failed container with a pure function.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess, IOFailure

>>> assert RequiresContextIOResult.from_value(1).alt(
...     lambda x: x + 1,
... )(...) == IOSuccess(1)

>>> assert RequiresContextIOResult.from_failure(1).alt(
...     lambda x: x + 1,
... )(...) == IOFailure(2)
Parameters

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

Return type

RequiresContextIOResult[-_EnvType, +_ValueType, ~_NewErrorType]

rescue(function)[source]

Composes this container with a function returning the same type.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess, IOFailure

>>> def rescuable(
...     arg: str,
... ) -> RequiresContextIOResult[str, str, str]:
...      if len(arg) > 1:
...          return RequiresContextIOResult(
...              lambda deps: IOSuccess(deps + arg),
...          )
...      return RequiresContextIOResult(
...          lambda deps: IOFailure(arg + deps),
...      )

>>> assert RequiresContextIOResult.from_value('a').rescue(
...     rescuable,
... )('c') == IOSuccess('a')
>>> assert RequiresContextIOResult.from_failure('a').rescue(
...     rescuable,
... )('c') == IOFailure('ac')
>>> assert RequiresContextIOResult.from_failure('aa').rescue(
...     rescuable,
... )('b') == IOSuccess('baa')
Parameters

function (Callable[[+_ErrorType], RequiresContextIOResult[-_EnvType, +_ValueType, ~_NewErrorType]]) –

Return type

RequiresContextIOResult[-_EnvType, +_ValueType, ~_NewErrorType]

value_or(default_value)[source]

Returns a callable that either returns a success or default value.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IO

>>> assert RequiresContextIOResult.from_value(1).value_or(2)(
...     RequiresContextIOResult.empty,
... ) == IO(1)

>>> assert RequiresContextIOResult.from_failure(1).value_or(2)(
...     RequiresContextIOResult.empty,
... ) == IO(2)
Parameters

default_value (~_FirstType) –

Return type

Callable[[-_EnvType], IO[Union[+_ValueType, ~_FirstType]]]

unwrap()[source]

Returns a callable that unwraps success value or raises exception.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IO

>>> assert RequiresContextIOResult.from_value(1).unwrap()(
...     RequiresContextIOResult.empty,
... ) == IO(1)

>>> RequiresContextIOResult.from_failure(1).unwrap()(
...     RequiresContextIOResult.empty,
... )
Traceback (most recent call last):
  ...
returns.primitives.exceptions.UnwrapFailedError
Return type

Callable[[-_EnvType], IO[+_ValueType]]

failure()[source]

Returns a callable that unwraps failure value or raises exception.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IO

>>> assert RequiresContextIOResult.from_failure(1).failure()(
...     RequiresContextIOResult.empty,
... ) == IO(1)

>>> RequiresContextIOResult.from_value(1).failure()(
...    RequiresContextIOResult.empty,
... )
Traceback (most recent call last):
  ...
returns.primitives.exceptions.UnwrapFailedError
Return type

Callable[[-_EnvType], IO[+_ErrorType]]

classmethod from_result(inner_value)[source]

Creates new container with Result as a unit value.

>>> from returns.context import RequiresContextIOResult
>>> from returns.result import Success, Failure
>>> from returns.io import IOSuccess, IOFailure
>>> deps = RequiresContextIOResult.empty

>>> assert RequiresContextIOResult.from_result(
...    Success(1),
... )(deps) == IOSuccess(1)

>>> assert RequiresContextIOResult.from_result(
...    Failure(1),
... )(deps) == IOFailure(1)
Parameters

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

Return type

RequiresContextIOResult[Any, +_ValueType, +_ErrorType]

classmethod from_io(inner_value)[source]

Creates new container from successful IO value.

>>> from returns.io import IO, IOSuccess
>>> from returns.context import RequiresContextIOResult

>>> assert RequiresContextIOResult.from_io(IO(1))(
...     RequiresContextIOResult.empty,
... ) == IOSuccess(1)
Parameters

inner_value (IO[~_NewValueType]) –

Return type

RequiresContextIOResult[Any, ~_NewValueType, Any]

classmethod from_failed_io(inner_value)[source]

Creates a new container from failed IO value.

>>> from returns.io import IO, IOFailure
>>> from returns.context import RequiresContextIOResult

>>> assert RequiresContextIOResult.from_failed_io(IO(1))(
...     RequiresContextIOResult.empty,
... ) == IOFailure(1)
Parameters

inner_value (IO[~_NewErrorType]) –

Return type

RequiresContextIOResult[Any, Any, ~_NewErrorType]

classmethod from_ioresult(inner_value)[source]

Creates new container with IOResult as a unit value.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess, IOFailure
>>> deps = RequiresContextIOResult.empty

>>> assert RequiresContextIOResult.from_ioresult(
...    IOSuccess(1),
... )(deps) == IOSuccess(1)

>>> assert RequiresContextIOResult.from_ioresult(
...    IOFailure(1),
... )(deps) == IOFailure(1)
Parameters

inner_value (IOResult[+_ValueType, +_ErrorType]) –

Return type

RequiresContextIOResult[Any, +_ValueType, +_ErrorType]

classmethod from_typecast(inner_value)[source]

You might end up with RequiresContext[IOResult] as a value.

This method is designed to turn it into RequiresContextIOResult. It will save all the typing information.

It is just more useful!

>>> from returns.context import RequiresContext
>>> from returns.io import IOSuccess, IOFailure

>>> assert RequiresContextIOResult.from_typecast(
...     RequiresContext.from_value(IOSuccess(1)),
... )(RequiresContextIOResult.empty) == IOSuccess(1)

>>> assert RequiresContextIOResult.from_typecast(
...     RequiresContext.from_value(IOFailure(1)),
... )(RequiresContextIOResult.empty) == IOFailure(1)
Parameters

inner_value (RequiresContext[_EnvType, IOResult[_NewValueType, _NewErrorType]]) –

Return type

RequiresContextIOResult[_EnvType, _NewValueType, _NewErrorType]

classmethod from_context(inner_value)[source]

Creates new container from RequiresContext as a success unit.

>>> from returns.context import RequiresContext
>>> from returns.io import IOSuccess

>>> assert RequiresContextIOResult.from_context(
...     RequiresContext.from_value(1),
... )(...) == IOSuccess(1)
Parameters

inner_value (RequiresContext[_EnvType, _FirstType]) –

Return type

RequiresContextIOResult[_EnvType, _FirstType, Any]

classmethod from_failed_context(inner_value)[source]

Creates new container from RequiresContext as a failure unit.

>>> from returns.context import RequiresContext
>>> from returns.io import IOFailure

>>> assert RequiresContextIOResult.from_failed_context(
...     RequiresContext.from_value(1),
... )(...) == IOFailure(1)
Parameters

inner_value (RequiresContext[_EnvType, _FirstType]) –

Return type

RequiresContextIOResult[_EnvType, Any, _FirstType]

classmethod from_result_context(inner_value)[source]

Creates new container from RequiresContextResult as a unit value.

>>> from returns.context import RequiresContextResult
>>> from returns.io import IOSuccess, IOFailure

>>> assert RequiresContextIOResult.from_result_context(
...     RequiresContextResult.from_value(1),
... )(...) == IOSuccess(1)

>>> assert RequiresContextIOResult.from_result_context(
...     RequiresContextResult.from_failure(1),
... )(...) == IOFailure(1)
Parameters

inner_value (RequiresContextResult[_EnvType, _ValueType, _ErrorType]) –

Return type

RequiresContextIOResult[_EnvType, _ValueType, _ErrorType]

classmethod from_value(inner_value)[source]

Creates new container with IOSuccess(inner_value) as a unit value.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess

>>> assert RequiresContextIOResult.from_value(1)(
...    RequiresContextIOResult.empty,
... ) == IOSuccess(1)
Parameters

inner_value (~_FirstType) –

Return type

RequiresContextIOResult[Any, ~_FirstType, Any]

classmethod from_failure(inner_value)[source]

Creates new container with IOFailure(inner_value) as a unit value.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOFailure

>>> assert RequiresContextIOResult.from_failure(1)(
...     RequiresContextIOResult.empty,
... ) == IOFailure(1)
Parameters

inner_value (~_FirstType) –

Return type

RequiresContextIOResult[Any, Any, ~_FirstType]

classmethod from_iterable(inner_value)[source]

Transforms an iterable of RequiresContextIOResult containers.

Returns a single container with multiple elements inside.

>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess, IOFailure

>>> assert RequiresContextIOResult.from_iterable([
...    RequiresContextIOResult.from_value(1),
...    RequiresContextIOResult.from_value(2),
... ])(...) == IOSuccess((1, 2))

>>> assert RequiresContextIOResult.from_iterable([
...    RequiresContextIOResult.from_value(1),
...    RequiresContextIOResult.from_failure('a'),
... ])(...) == IOFailure('a')

>>> assert RequiresContextIOResult.from_iterable([
...    RequiresContextIOResult.from_failure('a'),
...    RequiresContextIOResult.from_value(1),
... ])(...) == IOFailure('a')
Parameters

inner_value (Iterable[RequiresContextIOResult[-_EnvType, +_ValueType, +_ErrorType]]) –

Return type

RequiresContextIOResult[-_EnvType, Sequence[+_ValueType], +_ErrorType]

class ContextIOResult(*args, **kwds)[source]

Bases: returns.primitives.types.Immutable, typing.Generic

Helpers that can be used to work with RequiresContextIOResult container.

Related to returns.context.requires_context.Context and returns.context.requires_context_result.ContextResult, refer there for the docs.

classmethod ask()[source]

Is used to get the current dependencies inside the call stack.

Similar to returns.context.requires_context.Context.ask(), but returns IOResult instead of a regular value.

Please, refer to the docs there to learn how to use it.

One important note that is worth duplicating here: you might need to provide _EnvType explicitly, so mypy will know about it statically.

>>> from returns.context import ContextIOResult
>>> from returns.io import IOSuccess
>>> assert ContextIOResult[int].ask().map(str)(1) == IOSuccess('1')
Return type

RequiresContextIOResult[-_EnvType, -_EnvType, Any]

RequiresContextIOResultE

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

alias of returns.context.requires_context_ioresult.RequiresContextIOResult[_EnvType, _ValueType, Exception]

ReaderIOResult

Alias to save you some typing. Uses original name from Haskell.

alias of returns.context.requires_context_ioresult.RequiresContextIOResult[_EnvType, _ValueType, _ErrorType]

ReaderIOResultE

Alias to save you some typing. Uses Exception as error type.

alias of returns.context.requires_context_ioresult.RequiresContextIOResult[_EnvType, _ValueType, Exception]

RequiresContextFutureResult

graph TD; RequiresContextFutureResult ContextFutureResult BaseContainer --> RequiresContextFutureResult Generic --> RequiresContextFutureResult Immutable --> ContextFutureResult Generic --> ContextFutureResult
class RequiresContextFutureResult(inner_value)[source]

Bases: returns.primitives.container.BaseContainer, typing.Generic

The RequiresContextFutureResult combinator.

This probably the main type people are going to use in async programms.

See returns.context.requires_context.RequiresContext, returns.context.requires_context_result.RequiresContextResult, and returns.context.requires_context_result.RequiresContextIOResult for more docs.

This is just a handy wrapper around RequiresContext[env, FutureResult[a, b]] which represents a context-dependent impure operation that might fail.

So, this is a thin wrapper, without any changes in logic. Why do we need this wrapper? That’s just for better usability!

This way RequiresContextIOResult allows to simply work with:

  • raw values and pure functions

  • RequiresContext values and pure functions returning it

  • RequiresContextResult values and pure functions returning it

  • RequiresContextIOResult values and pure functions returning it

  • Result and pure functions returning it

  • IOResult and functions returning it

  • FutureResult and functions returning it

  • other RequiresContextFutureResult related functions and values

This is a complex type for complex tasks!

Important implementation detail: due it is meaning, RequiresContextFutureResult cannot have Success and Failure subclasses.

We only have just one type. That’s by design.

Different converters are also not supported for this type. Use converters inside the RequiresContext context, not outside.

Parameters

inner_value (Callable[[-_EnvType], FutureResult[+_ValueType, +_ErrorType]]) –

empty: ClassVar[Any] = <object object>

A convinient placeholder to call methods created by .from_value().

map(function)[source]

Composes successful container with a pure function.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.io import IOSuccess, IOFailure

>>> assert anyio.run(RequiresContextFutureResult.from_value(1).map(
...     lambda x: x + 1,
... )(...).awaitable) == IOSuccess(2)

>>> assert anyio.run(RequiresContextFutureResult.from_failure(1).map(
...     lambda x: x + 1,
... )(...).awaitable) == IOFailure(1)
Parameters

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

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

apply(container)[source]

Calls a wrapped function in a container on this container.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.io import IOSuccess, IOFailure

>>> def transform(arg: str) -> str:
...     return arg + 'b'

>>> assert anyio.run(
...    RequiresContextFutureResult.from_value('a').apply(
...        RequiresContextFutureResult.from_value(transform),
...    ),
...    RequiresContextFutureResult.empty,
... ) == IOSuccess('ab')

>>> assert anyio.run(
...    RequiresContextFutureResult.from_failure('a').apply(
...        RequiresContextFutureResult.from_value(transform),
...    ),
...    RequiresContextFutureResult.empty,
... ) == IOFailure('a')
Parameters

container (RequiresContextFutureResult[-_EnvType, Callable[[+_ValueType], ~_NewValueType], +_ErrorType]) –

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind(function)[source]

Composes this container with a function returning the same type.

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

>>> def function(
...     number: int,
... ) -> RequiresContextFutureResult[int, str, int]:
...     # `deps` has `int` type here:
...     return RequiresContextFutureResult(
...         lambda deps: FutureResult.from_value(str(number + deps)),
...     )

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(2).bind(function),
...     3,
... ) == IOSuccess('5')
>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure(2).bind(function),
...     3,
... ) == IOFailure(2)
Parameters

function (Callable[[+_ValueType], RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]]) –

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_async(function)[source]

Composes this container with a async function returning the same type.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.io import IOSuccess, IOFailure

>>> async def function(
...     number: int,
... ) -> RequiresContextFutureResult[int, str, int]:
...     return RequiresContextFutureResult.from_value(number + 1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).bind_async(
...        function,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(2)
>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure(1).bind_async(
...        function,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

function (Callable[[+_ValueType], Awaitable[RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]]]) –

Return type

RequiresContextFutureResult[-_EnvType, ~_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.context import RequiresContextFutureResult
>>> from returns.io import IOSuccess, IOFailure

>>> async def coroutine(x: int) -> int:
...    return x + 1

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).bind_awaitable(
...         coroutine,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(2)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure(1).bind_awaitable(
...         coroutine,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

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

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_result(function)[source]

Binds Result returning function to the current container.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.result import Success, Result
>>> from returns.io import IOSuccess, IOFailure

>>> def function(num: int) -> Result[int, str]:
...     return Success(num + 1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).bind_result(
...         function,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(2)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure(':(').bind_result(
...         function,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(':(')
Parameters

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

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_context(function)[source]

Binds RequiresContext returning function to current container.

>>> import anyio
>>> from returns.context import RequiresContext
>>> from returns.io import IOSuccess, IOFailure

>>> def function(arg: int) -> RequiresContext[str, int]:
...     return RequiresContext(lambda deps: len(deps) + arg)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(2).bind_context(
...         function,
...     ),
...     'abc',
... ) == IOSuccess(5)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure(0).bind_context(
...         function,
...     ),
...     'abc',
... ) == IOFailure(0)
Parameters

function (Callable[[+_ValueType], ForwardRef]) –

Return type

RequiresContextFutureResult[_EnvType, _NewValueType, _ErrorType]

bind_context_result(function)[source]

Binds RequiresContextResult returning function to the current one.

>>> import anyio
>>> from returns.context import RequiresContextResult
>>> from returns.io import IOSuccess, IOFailure
>>> from returns.result import Success

>>> def function(arg: int) -> RequiresContextResult[str, int, int]:
...     return RequiresContextResult(
...         lambda deps: Success(len(deps) + arg),
...     )

>>> instance = RequiresContextFutureResult.from_value(
...    2,
... ).bind_context_result(
...     function,
... )('abc')
>>> assert anyio.run(instance.awaitable) == IOSuccess(5)

>>> instance = RequiresContextFutureResult.from_failure(
...    2,
... ).bind_context_result(
...     function,
... )('abc')
>>> assert anyio.run(instance.awaitable) == IOFailure(2)
Parameters

function (Callable[[+_ValueType], ForwardRef]) –

Return type

RequiresContextFutureResult[_EnvType, _NewValueType, _ErrorType]

bind_context_ioresult(function)[source]

Binds RequiresContextIOResult returning function to the current one.

>>> import anyio
>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess, IOFailure

>>> def function(arg: int) -> RequiresContextIOResult[str, int, int]:
...     return RequiresContextIOResult(
...         lambda deps: IOSuccess(len(deps) + arg),
...     )

>>> instance = RequiresContextFutureResult.from_value(
...    2,
... ).bind_context_ioresult(
...     function,
... )('abc')
>>> assert anyio.run(instance.awaitable) == IOSuccess(5)

>>> instance = RequiresContextFutureResult.from_failure(
...    2,
... ).bind_context_ioresult(
...     function,
... )('abc')
>>> assert anyio.run(instance.awaitable) == IOFailure(2)
Parameters

function (Callable[[+_ValueType], ForwardRef]) –

Return type

RequiresContextFutureResult[_EnvType, _NewValueType, _ErrorType]

bind_io(function)[source]

Binds IO returning function to the current container.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.io import IO, IOSuccess, IOFailure

>>> def do_io(number: int) -> IO[str]:
...     return IO(str(number))  # not IO operation actually

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).bind_io(do_io),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess('1')

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure(1).bind_io(do_io),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

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

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_ioresult(function)[source]

Binds IOResult returning function to the current container.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.io import IOResult, IOSuccess, IOFailure

>>> def function(num: int) -> IOResult[int, str]:
...     return IOSuccess(num + 1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).bind_ioresult(
...         function,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(2)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure(':(').bind_ioresult(
...         function,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(':(')
Parameters

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

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_future(function)[source]

Binds Future returning function to the current container.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.future import Future
>>> from returns.io import IOSuccess, IOFailure

>>> def function(num: int) -> Future[int]:
...     return Future.from_value(num + 1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).bind_future(
...         function,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(2)

>>> failed = RequiresContextFutureResult.from_failure(':(')
>>> assert anyio.run(
...     failed.bind_future(function),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(':(')
Parameters

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

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_future_result(function)[source]

Binds FutureResult returning function to the current container.

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

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

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).bind_future_result(
...         function,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(2)

>>> failed = RequiresContextFutureResult.from_failure(':(')
>>> assert anyio.run(
...     failed.bind_future_result(function),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(':(')
Parameters

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

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_async_future(function)[source]

Binds Future returning async function to the current container.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.future import Future
>>> from returns.io import IOSuccess, IOFailure

>>> async def function(num: int) -> Future[int]:
...     return Future.from_value(num + 1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).bind_async_future(
...         function,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(2)

>>> failed = RequiresContextFutureResult.from_failure(':(')
>>> assert anyio.run(
...     failed.bind_async_future(function),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(':(')
Parameters

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

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

bind_async_future_result(function)[source]

Bind FutureResult returning async function to the current container.

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

>>> async def function(num: int) -> FutureResult[int, str]:
...     return FutureResult.from_value(num + 1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(
...         1,
...     ).bind_async_future_result(
...         function,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(2)

>>> failed = RequiresContextFutureResult.from_failure(':(')
>>> assert anyio.run(
...     failed.bind_async_future_result(function),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(':(')
Parameters

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

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

fix(function)[source]

Composes failed container with a pure function.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.io import IOSuccess

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).fix(
...        lambda x: x + 1,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure(1).fix(
...        lambda x: x + 1,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(2)
Parameters

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

Return type

RequiresContextFutureResult[-_EnvType, ~_NewValueType, +_ErrorType]

alt(function)[source]

Composes failed container with a pure function.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.io import IOSuccess, IOFailure

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).alt(
...        lambda x: x + 1,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure(1).alt(
...        lambda x: x + 1,
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(2)
Parameters

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

Return type

RequiresContextFutureResult[-_EnvType, +_ValueType, ~_NewErrorType]

rescue(function)[source]

Composes this container with a function returning the same type.

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

>>> def rescuable(
...     arg: str,
... ) -> RequiresContextFutureResult[str, str, str]:
...      return RequiresContextFutureResult(
...          lambda deps: FutureResult.from_value(
...              deps + arg,
...          ),
...      )

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value('a').rescue(rescuable),
...     'c',
... ) == IOSuccess('a')

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure('aa').rescue(
...         rescuable,
...     ),
...     'b',
... ) == IOSuccess('baa')
Parameters

function (Callable[[+_ErrorType], RequiresContextFutureResult[-_EnvType, +_ValueType, ~_NewErrorType]]) –

Return type

RequiresContextFutureResult[-_EnvType, +_ValueType, ~_NewErrorType]

value_or(default_value)[source]

Returns a callable that either returns a success or default value.

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

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).value_or(2),
...     RequiresContextFutureResult.empty,
... ) == IO(1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure(1).value_or(2),
...     RequiresContextFutureResult.empty,
... ) == IO(2)
Parameters

default_value (~_FirstType) –

Return type

Callable[[-_EnvType], Awaitable[IO[Union[+_ValueType, ~_FirstType]]]]

unwrap()[source]

Returns a callable that unwraps success value or raises exception.

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

>>> assert anyio.run(
...     RequiresContextFutureResult.from_value(1).unwrap(),
...     RequiresContextFutureResult.empty,
... ) == IO(1)

>>> anyio.run(
...     RequiresContextFutureResult.from_failure(1).unwrap(),
...     RequiresContextFutureResult.empty,
... )
Traceback (most recent call last):
  ...
returns.primitives.exceptions.UnwrapFailedError
Return type

Callable[[-_EnvType], Awaitable[IO[+_ValueType]]]

failure()[source]

Returns a callable that unwraps failure value or raises exception.

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

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failure(1).failure(),
...     RequiresContextFutureResult.empty,
... ) == IO(1)

>>> anyio.run(
...    RequiresContextFutureResult.from_value(1).failure(),
...    RequiresContextFutureResult.empty,
... )
Traceback (most recent call last):
  ...
returns.primitives.exceptions.UnwrapFailedError
Return type

Callable[[-_EnvType], Awaitable[IO[+_ErrorType]]]

classmethod from_result(inner_value)[source]

Creates new container with Result as a unit value.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.result import Success, Failure
>>> from returns.io import IOSuccess, IOFailure

>>> assert anyio.run(
...    RequiresContextFutureResult.from_result(Success(1)),
...    RequiresContextFutureResult.empty,
... ) == IOSuccess(1)

>>> assert anyio.run(
...    RequiresContextFutureResult.from_result(Failure(1)),
...    RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

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

Return type

RequiresContextFutureResult[Any, +_ValueType, +_ErrorType]

classmethod from_io(inner_value)[source]

Creates new container from successful IO value.

>>> import anyio
>>> from returns.io import IO, IOSuccess
>>> from returns.context import RequiresContextFutureResult

>>> assert anyio.run(
...     RequiresContextFutureResult.from_io(IO(1)),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(1)
Parameters

inner_value (IO[~_NewValueType]) –

Return type

RequiresContextFutureResult[Any, ~_NewValueType, Any]

classmethod from_failed_io(inner_value)[source]

Creates a new container from failed IO value.

>>> import anyio
>>> from returns.io import IO, IOFailure
>>> from returns.context import RequiresContextFutureResult

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failed_io(IO(1)),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

inner_value (IO[~_NewErrorType]) –

Return type

RequiresContextFutureResult[Any, Any, ~_NewErrorType]

classmethod from_ioresult(inner_value)[source]

Creates new container with IOResult as a unit value.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.io import IOSuccess, IOFailure

>>> assert anyio.run(
...    RequiresContextFutureResult.from_ioresult(IOSuccess(1)),
...    RequiresContextFutureResult.empty,
... ) == IOSuccess(1)

>>> assert anyio.run(
...    RequiresContextFutureResult.from_ioresult(IOFailure(1)),
...    RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

inner_value (IOResult[+_ValueType, +_ErrorType]) –

Return type

RequiresContextFutureResult[Any, +_ValueType, +_ErrorType]

classmethod from_future(inner_value)[source]

Creates new container with successful Future as a unit value.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.future import Future
>>> from returns.io import IOSuccess

>>> assert anyio.run(
...    RequiresContextFutureResult.from_future(Future.from_value(1)),
...    RequiresContextFutureResult.empty,
... ) == IOSuccess(1)
Parameters

inner_value (Future[+_ValueType]) –

Return type

RequiresContextFutureResult[Any, +_ValueType, Any]

classmethod from_failed_future(inner_value)[source]

Creates new container with failed Future as a unit value.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.future import Future
>>> from returns.io import IOFailure

>>> assert anyio.run(
...    RequiresContextFutureResult.from_failed_future(
...        Future.from_value(1),
...    ),
...    RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

inner_value (Future[+_ErrorType]) –

Return type

RequiresContextFutureResult[Any, Any, +_ErrorType]

classmethod from_future_result(inner_value)[source]

Creates new container with FutureResult as a unit value.

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

>>> assert anyio.run(
...    RequiresContextFutureResult.from_future_result(
...        FutureResult.from_value(1),
...    ),
...    RequiresContextFutureResult.empty,
... ) == IOSuccess(1)

>>> assert anyio.run(
...    RequiresContextFutureResult.from_future_result(
...        FutureResult.from_failure(1),
...    ),
...    RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

inner_value (FutureResult[+_ValueType, +_ErrorType]) –

Return type

RequiresContextFutureResult[Any, +_ValueType, +_ErrorType]

classmethod from_typecast(inner_value)[source]

You might end up with RequiresContext[FutureResult] as a value.

This method is designed to turn it into RequiresContextFutureResult. It will save all the typing information.

It is just more useful!

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

>>> assert anyio.run(
...     RequiresContextFutureResult.from_typecast(
...         RequiresContext.from_value(FutureResult.from_value(1)),
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_typecast(
...         RequiresContext.from_value(FutureResult.from_failure(1)),
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

inner_value (RequiresContext[_EnvType, FutureResult[_NewValueType, _NewErrorType]]) –

Return type

RequiresContextFutureResult[_EnvType, _NewValueType, _NewErrorType]

classmethod from_context(inner_value)[source]

Creates new container from RequiresContext as a success unit.

>>> import anyio
>>> from returns.context import RequiresContext
>>> from returns.io import IOSuccess

>>> assert anyio.run(
...     RequiresContextFutureResult.from_context(
...         RequiresContext.from_value(1),
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(1)
Parameters

inner_value (RequiresContext[_EnvType, _FirstType]) –

Return type

RequiresContextFutureResult[_EnvType, _FirstType, Any]

classmethod from_failed_context(inner_value)[source]

Creates new container from RequiresContext as a failure unit.

>>> import anyio
>>> from returns.context import RequiresContext
>>> from returns.io import IOFailure

>>> assert anyio.run(
...     RequiresContextFutureResult.from_failed_context(
...         RequiresContext.from_value(1),
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

inner_value (RequiresContext[_EnvType, _FirstType]) –

Return type

RequiresContextFutureResult[_EnvType, Any, _FirstType]

classmethod from_result_context(inner_value)[source]

Creates new container from RequiresContextResult as a unit value.

>>> import anyio
>>> from returns.context import RequiresContextResult
>>> from returns.io import IOSuccess, IOFailure

>>> assert anyio.run(
...     RequiresContextFutureResult.from_result_context(
...         RequiresContextResult.from_value(1),
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_result_context(
...         RequiresContextResult.from_failure(1),
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

inner_value (RequiresContextResult[_EnvType, _ValueType, _ErrorType]) –

Return type

RequiresContextFutureResult[_EnvType, _ValueType, _ErrorType]

classmethod from_ioresult_context(inner_value)[source]

Creates new container from RequiresContextIOResult as a unit value.

>>> import anyio
>>> from returns.context import RequiresContextIOResult
>>> from returns.io import IOSuccess, IOFailure

>>> assert anyio.run(
...     RequiresContextFutureResult.from_ioresult_context(
...         RequiresContextIOResult.from_value(1),
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOSuccess(1)

>>> assert anyio.run(
...     RequiresContextFutureResult.from_ioresult_context(
...         RequiresContextIOResult.from_failure(1),
...     ),
...     RequiresContextFutureResult.empty,
... ) == IOFailure(1)
Parameters

inner_value (RequiresContextIOResult[_EnvType, _ValueType, _ErrorType]) –

Return type

RequiresContextFutureResult[_EnvType, _ValueType, _ErrorType]

classmethod from_value(inner_value)[source]

Creates new container with successful FutureResult as a unit value.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.io import IOSuccess

>>> assert anyio.run(RequiresContextFutureResult.from_value(1)(
...    RequiresContextFutureResult.empty,
... ).awaitable) == IOSuccess(1)
Parameters

inner_value (~_FirstType) –

Return type

RequiresContextFutureResult[Any, ~_FirstType, Any]

classmethod from_failure(inner_value)[source]

Creates new container with failed FutureResult as a unit value.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.io import IOFailure

>>> assert anyio.run(RequiresContextFutureResult.from_failure(1)(
...     RequiresContextFutureResult.empty,
... ).awaitable) == IOFailure(1)
Parameters

inner_value (~_FirstType) –

Return type

RequiresContextFutureResult[Any, Any, ~_FirstType]

classmethod from_iterable(inner_value)[source]

Transforms an iterable of RequiresContextFutureResult containers.

Returns a single container with multiple elements inside.

>>> import anyio
>>> from returns.context import RequiresContextFutureResult
>>> from returns.io import IOSuccess, IOFailure

>>> assert anyio.run(
...    RequiresContextFutureResult.from_iterable([
...        RequiresContextFutureResult.from_value(1),
...        RequiresContextFutureResult.from_value(2),
...    ]),
...    RequiresContextFutureResult.empty,
... ) == IOSuccess((1, 2))

>>> assert anyio.run(
...    RequiresContextFutureResult.from_iterable([
...        RequiresContextFutureResult.from_value(1),
...        RequiresContextFutureResult.from_failure('a'),
...    ]),
...    RequiresContextFutureResult.empty,
... ) == IOFailure('a')
Parameters

inner_value (Iterable[RequiresContextFutureResult[-_EnvType, +_ValueType, +_ErrorType]]) –

Return type

RequiresContextFutureResult[-_EnvType, Sequence[+_ValueType], +_ErrorType]

class ContextFutureResult(*args, **kwds)[source]

Bases: returns.primitives.types.Immutable, typing.Generic

Helpers that can be used to work with ReaderFutureResult container.

Related to returns.context.requires_context.Context and returns.context.requires_context_result.ContextResult, refer there for the docs.

classmethod ask()[source]

Is used to get the current dependencies inside the call stack.

Similar to returns.context.requires_context.Context.ask(), but returns IOResult instead of a regular value.

Please, refer to the docs there to learn how to use it.

One important note that is worth duplicating here: you might need to provide _EnvType explicitly, so mypy will know about it statically.

>>> import anyio
>>> from returns.context import ContextFutureResult
>>> from returns.io import IOSuccess

>>> assert anyio.run(
...     ContextFutureResult[int].ask().map(str),
...     1,
... ) == IOSuccess('1')
Return type

RequiresContextFutureResult[-_EnvType, -_EnvType, Any]

RequiresContextFutureResultE

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

alias of returns.context.requires_context_future_result.RequiresContextFutureResult[_EnvType, _ValueType, Exception]

ReaderFutureResult

Sometimes RequiresContextFutureResult is too long to type.

Parameters

inner_value (Callable[[-_EnvType], FutureResult[+_ValueType, +_ErrorType]]) –

alias of returns.context.requires_context_future_result.RequiresContextFutureResult

ReaderFutureResultE

Alias to save you some typing. Uses Exception as error type.

alias of returns.context.requires_context_future_result.RequiresContextFutureResult[_EnvType, _ValueType, Exception]

Future Pipelines