Pointfree

This module provides a bunch of primitives to work with containers.

It is centered around the composition idea. Sometimes using methods on containers is not very helpful. Instead we can use functions that has the reverse semantics, but the same end result.

Why would anyone need these functions when you can use methods? To create pipelines!

Without pointfree functions you cannot easily work with containers inside pipelines. Because they do not compose well:

from returns.pipeline import pipe
from returns.result import ResultE

def returns_result(arg: int) -> ResultE[int]:
    ...

def works_with_result(arg: int) -> ResultE[int]:
    ...

def finish_work(arg: int) -> ResultE[int]:
    ...

pipe(
    returns_result,
    works_with_result,  # does not compose!
    finish_work,  # does not compose either!
)

In a normal situation you would probably write:

returns_result().bind(works_with_result).bind(notifies_user)

And you need a way to somehow do this in the pipeline. That’s where pointfree functions become really useful.

map

Allows to compose cointainers and functions, but in a reverse manner.

>>> from returns.pointfree import map_
>>> from returns.maybe import Maybe, Some

>>> def mappable(arg: str) -> int:
...     return ord(arg)

>>> container: Maybe[str] = Some('a')
>>> # We now have two way of composining these entities.
>>> # 1. Via ``.map``:
>>> assert container.map(mappable) == Some(97)
>>> # 2. Or via ``bind`` function, the same but in the inverse way:
>>> assert map_(mappable)(container) == Some(97)

bind

Allows to bind a function that returns a container of the same type.

Without bind() function it would be very hard to declaratively compose two entities:

  1. Existings container

  2. Existing functions that accepts a regular value and returns a container

We can compose these entities with .bind when calling it directly, but how can we do it inversevely?

>>> from returns.pointfree import bind
>>> from returns.maybe import Maybe, Some

>>> def bindable(arg: str) -> Maybe[int]:
...     return Some(1)

>>> container: Maybe[str] = Some('a')
>>> # We now have two way of composining these entities.
>>> # 1. Via ``.bind``:
>>> assert container.bind(bindable) == Some(1)
>>> # 2. Or via ``bind`` function, the same but in the inverse way:
>>> assert bind(bindable)(container) == Some(1)

That’s it!

We also have a long list of other bind_* functions, like:

  • bind_io to bind functions returning IO container

  • bind_result to bind functions returning Result container

  • bind_ioresult to bind functions returning IOResult container

  • bind_future to bind functions returning Future container

  • bind_async_future to bind async functions returning Future container

  • bind_future_result to bind functions returning FutureResult container

  • bind_async_future_result to bind async functions returning FutureResult container

  • bind_context to bind functions returning RequiresContext container

  • bind_context_result to bind functions returning RequiresContextResult container

  • bind_context_ioresult to bind functions returning RequiresContextIOResult container

  • bind_async to bind async functions returning Future or FutureResult

  • bind_awaitable to bind async non-container functions

unify

unify is the same as bind, but returns different return type. While bind reveals strictly the same error type, unify returns Union[_OldReturnType, _NewReturnType].

rescue

Pointfree rescue() function is an alternative to .rescue() container method. It is also required for better declarative programming.

>>> from returns.pointfree import rescue
>>> from returns.result import Success, Failure, Result

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

>>> container: Result[int, str] = Failure('a')
>>> # We now have two way of composining these entities.
>>> # 1. Via ``.rescue``:
>>> assert container.rescue(function) == Success(1)
>>> # 2. Or via ``rescue`` function, the same but in the inverse way:
>>> assert rescue(function)(container) == Success(1)

apply

Pointfree apply function allows to use .apply() container method like a function:

>>> from returns.pointfree import apply
>>> from returns.maybe import Some, Nothing

>>> def function(arg: int) -> str:
...     return chr(arg) + '!'

>>> assert apply(Some(function))(Some(97)) == Some('a!')
>>> assert apply(Some(function))(Some(98)) == Some('b!')
>>> assert apply(Some(function))(Nothing) == Nothing
>>> assert apply(Nothing)(Nothing) == Nothing

If you wish to use apply inside a pipeline that’s how it would probably look like:

>>> from returns.pointfree import apply
>>> from returns.pipeline import flow
>>> from returns.maybe import Some

>>> def function(arg: int) -> str:
...     return chr(arg) + '!'

>>> assert flow(
...     Some(97),
...     apply(Some(function)),
... ) == Some('a!')

Or with function as the first parameter:

>>> from returns.pipeline import flow
>>> from returns.curry import curry
>>> from returns.maybe import Some

>>> @curry
... def function(first: int, second: int) -> int:
...     return first + second

>>> assert flow(
...     Some(function),
...     Some(2).apply,
...     Some(3).apply,
... ) == Some(5)

value_or

Allows to unwrap values from containers:

>>> from returns.pointfree import value_or
>>> from returns.result import Result
>>> from returns.pipeline import flow

>>> assert flow(Result.from_value(1), value_or(2)) == 1
>>> assert flow(Result.from_failure(1), value_or(2)) == 2

All containers with .value_or method are supported.

API Reference

map_(function)

Lifts function to be wrapped in a container for better composition.

In other words, it modifies the function’s signature from: a -> b to: Container[a] -> Container[b]

Works similar to returns.primitives.interfaces.Mappable.map(), but has inverse semantics.

This is how it should be used:

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

>>> def example(argument: int) -> float:
...     return argument / 2  # not a container!

>>> async def main() -> Future[float]:
...     return await map_(example)(Future.from_value(1))

>>> assert anyio.run(main) == IO(0.5)

See also

bind(function)

Turns function’s input parameter from a regular value to a container.

In other words, it modifies the function signature from: a -> Container[b] to: Container[a] -> Container[b]

Similar to returns.pointfree.rescue(), but works for successful containers.

This is how it should be used:

>>> from returns.pointfree import bind
>>> from returns.maybe import Maybe, Some, Nothing

>>> def example(argument: int) -> Maybe[int]:
...     return Some(argument + 1)

>>> assert bind(example)(Some(1)) == Some(2)
>>> assert bind(example)(Nothing) == Nothing

Note, that this function works for all containers with .bind method. See returns.primitives.interfaces.Bindable for more info.

bind_result(function)

Composes successful container with a function that returns a container.

In other words, it modifies the function’s signature from: a -> Result[b, c] to: Container[a, c] -> Container[b, c]

>>> from returns.io import IOSuccess
>>> from returns.context import RequiresContextResult
>>> from returns.result import Result, Success
>>> from returns.pointfree import bind_result

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

>>> bound = bind_result(returns_result)
>>> assert bound(IOSuccess(1)) == IOSuccess(2)
>>> assert bound(RequiresContextResult.from_value(1))(...) == Success(2)
bind_io(function)

Lifts IO function to be wrapped in other container.

In other words, it modifies the function’s signature from: a -> IO[b] to: Container[a] -> Container[b]

This is how it should be used:

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

>>> def example(argument: int) -> IO[float]:
...     return IO(argument / 2)

>>> async def main() -> Future[float]:
...     container = Future.from_value(1)
...     return await bind_io(example)(container)

>>> assert anyio.run(main) == IO(0.5)

Or with sync code:

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

>>> def returns_io(arg: int) -> IO[float]:
...     return IO(arg + 0.5)

>>> bound = bind_io(returns_io)
>>> assert bound(IOSuccess(1)) == IOSuccess(1.5)
>>> assert bound(IOFailure(1)) == IOFailure(1)
bind_ioresult(function)

Lifts function returning IOResult to be wrapped in another container.

In other words, it modifies the function’s signature from: a -> IOResult[b, c] to: Container[a, c] -> Container[b, c]

This is how it should be used:

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

>>> def example(argument: int) -> IOResult[float, int]:
...     return IOSuccess(argument / 2)

>>> async def success() -> FutureResult[float, int]:
...     container = FutureResult.from_value(1)
...     return await bind_ioresult(example)(container)

>>> async def failure() -> FutureResult[float, int]:
...     container = FutureResult.from_failure(1)
...     return await bind_ioresult(example)(container)

>>> assert anyio.run(success) == IOSuccess(0.5)
>>> assert anyio.run(failure) == IOFailure(1)

And with sync code:

>>> from returns.context import RequiresContextIOResult

>>> def function(arg: int) -> IOResult[str, int]:
...     if arg > 0:
...         return IOSuccess(str(arg) + '!')
...     return IOFailure(arg)

>>> deps = RequiresContextIOResult.empty

>>> assert bind_ioresult(function)(
...     RequiresContextIOResult.from_value(1),
... )(deps) == IOSuccess('1!')

>>> assert bind_ioresult(function)(
...     RequiresContextIOResult.from_value(0),
... )(deps) == IOFailure(0)

>>> assert bind_ioresult(function)(
...     RequiresContextIOResult.from_failure('nope'),
... )(deps) == IOFailure('nope')
bind_future(function)

Lifts Future function to be wrapped in other container.

In other words, it modifies the function’s signature from: a -> Future[b] to: Container[a] -> Container[b]

This is how it should be used:

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

>>> def example(argument: int) -> Future[float]:
...     return Future.from_value(argument / 2)

>>> async def success() -> FutureResult[float, int]:
...     container = FutureResult.from_value(1)
...     return await bind_future(example)(container)

>>> async def failure() -> FutureResult[float, int]:
...     container = FutureResult.from_failure(1)
...     return await bind_future(example)(container)

>>> assert anyio.run(success) == IOSuccess(0.5)
>>> assert anyio.run(failure) == IOFailure(1)
bind_async_future(function)

Compose a container and async function returning a container.

In other words, it modifies the function’s signature from: a -> Awaitable[Container[b]] to: Container[a] -> Container[b]

This is how it should be used:

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

>>> async def coroutine(x: int) -> Future[str]:
...    return Future.from_value(str(x + 1))

>>> bound = bind_async_future(coroutine)(FutureResult.from_value(1))
>>> assert anyio.run(bound.awaitable) == IOSuccess('2')

>>> bound = bind_async_future(coroutine)(FutureResult.from_failure(1))
>>> assert anyio.run(bound.awaitable) == IOFailure(1)
bind_future_result(function)

Lifts FutureResult function to be wrapped in other container.

In other words, it modifies the function’s signature from: a -> FutureResult[b, c] to: Container[a, c] -> Container[b, c]

This is how it should be used:

>>> import anyio
>>> from returns.future import FutureResultE
>>> from returns.context import ReaderFutureResultE
>>> from returns.io import IOSuccess, IOFailure
>>> from returns.pointfree import bind_future_result

>>> def example(argument: int) -> FutureResultE[float]:
...     return FutureResultE.from_value(argument / 2)

>>> assert anyio.run(bind_future_result(example)(
...     ReaderFutureResultE.from_value(1),
... ), ReaderFutureResultE.empty) == IOSuccess(0.5)

>>> assert anyio.run(bind_future_result(example)(
...     ReaderFutureResultE.from_failure(':('),
... ), ReaderFutureResultE.empty) == IOFailure(':(')
bind_async_future_result(function)

Compose a container and async function returning a FutureResult.

In other words, it modifies the function’s signature from: a -> Awaitable[FutureResult[b, c]] to: Container[a, c] -> Container[b, c]

This is how it should be used:

>>> import anyio
>>> from returns.future import FutureResult
>>> from returns.context import ReaderFutureResult
>>> from returns.io import IOSuccess, IOFailure
>>> from returns.pointfree import bind_async_future_result

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

>>> bound = bind_async_future_result(coroutine)(
...     ReaderFutureResult.from_value(1),
... )
>>> assert anyio.run(bound, ReaderFutureResult.empty) == IOSuccess('2')

>>> bound = bind_async_future_result(coroutine)(
...     ReaderFutureResult.from_failure(1),
... )
>>> assert anyio.run(bound, ReaderFutureResult.empty) == IOFailure(1)
bind_context(function)

Lifts function from RequiresContext for better composition.

In other words, it modifies the function’s signature from: a -> RequiresContext[env, b] to: Container[env, a, c] -> Container[env, b, c]

>>> from returns.context import RequiresContext, RequiresContextResult
>>> from returns.result import Success, Failure
>>> from returns.pointfree import bind_context

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

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

>>> assert bind_context(function)(
...     RequiresContextResult.from_failure(0),
... )('abc') == Failure(0)
bind_context_result(function)

Lifts function from RequiresContextResult for better composition.

In other words, it modifies the function’s signature from: a -> RequiresContextResult[env, b, c] to: Container[env, a, c] -> Container[env, b, c]

>>> from returns.context import (
...     RequiresContextResult,
...     RequiresContextIOResult,
... )
>>> from returns.io import IOSuccess, IOFailure
>>> from returns.result import Success
>>> from returns.pointfree import bind_context_result

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

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

>>> assert bind_context_result(function)(
...     RequiresContextIOResult.from_failure(0),
... )('abc') == IOFailure(0)
bind_context_ioresult(function)

Lifts function from RequiresContextIOResult for better composition.

In other words, it modifies the function’s signature from: a -> RequiresContextResult[env, b, c] to: Container[env, a, c] -> Container[env, b, c]

>>> import anyio
>>> from returns.context import (
...     RequiresContextFutureResult,
...     RequiresContextIOResult,
... )
>>> from returns.io import IOSuccess, IOFailure
>>> from returns.pointfree import bind_context_ioresult

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

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

>>> assert anyio.run(bind_context_ioresult(function)(
...     RequiresContextFutureResult.from_failure(0),
... )('abc').awaitable) == IOFailure(0)
bind_async(function)

Compose a container and async function returning a container.

In other words, it modifies the function’s signature from: a -> Awaitable[Container[b]] to: Container[a] -> Container[b]

This is how it should be used:

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

>>> async def coroutine(x: int) -> Future[str]:
...    return Future.from_value(str(x + 1))

>>> bound = bind_async(coroutine)(Future.from_value(1))
>>> assert anyio.run(bound.awaitable) == IO('2')
bind_awaitable(function)

Composes a container a regular async function.

This function should return plain, non-container value.

In other words, it modifies the function’s signature from: a -> Awaitable[b] to: Container[a] -> Container[b]

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

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

>>> assert anyio.run(
...     bind_awaitable(coroutine)(Future.from_value(1)).awaitable,
... ) == IO(2)
unify(function)

Turns function’s input parameter from a regular value to a container.

In other words, it modifies the function signature from: a -> Container[b] to: Container[a] -> Container[b]

Similar to returns.pointfree.bind(), but unifies the result error type to become Union[_ErrorType, _NewErrorType].

Similar to returns.pointfree.rescue(), but works for successful containers.

This is how it should be used:

>>> from returns.pointfree import unify
>>> from returns.result import Success, Result, Failure

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

>>> assert unify(example)(Success(1)) == Success(2)
>>> assert unify(example)(Failure('a')) == Failure('a')

Note, that this function works for all containers with .unify method. See returns.primitives.interfaces.Bindable for more info.

rescue(function)

Turns function’s input parameter from a regular value to a container.

In other words, it modifies the function signature from: a -> Container[b] to: Container[a] -> Container[b]

Similar to returns.pointfree.bind(), but works for failed containers.

This is how it should be used:

>>> from returns.pointfree import rescue
>>> from returns.result import Success, Failure, Result

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

>>> assert rescue(example)(Success('a')) == Success('a')
>>> assert rescue(example)(Failure(1)) == Success(2)

Note, that this function works for all containers with .rescue method. See returns.primitives.interfaces.Rescueable for more info.

apply(container)

Turns container containing a function into a callable.

In other words, it modifies the function signature from: Container[a -> b] to: Container[a] -> Container[b]

This is how it should be used:

>>> from returns.pointfree import apply
>>> from returns.maybe import Some, Nothing

>>> def example(argument: int) -> int:
...     return argument + 1

>>> assert apply(Some(example))(Some(1)) == Some(2)
>>> assert apply(Some(example))(Nothing) == Nothing
>>> assert apply(Nothing)(Some(1)) == Nothing
>>> assert apply(Nothing)(Nothing) == Nothing

Note, that this function works for all containers with .apply method. See returns.primitives.interfaces.Applicative for more info.

value_or(default_value)

Get value from successful container or default value from failed one.

>>> import anyio
>>> from returns.pointfree import value_or
>>> from returns.io import IO, IOFailure, IOSuccess
>>> from returns.maybe import Some, Nothing
>>> from returns.context import ReaderFutureResult

>>> assert value_or(2)(IOSuccess(1)) == IO(1)
>>> assert value_or(2)(IOFailure(1)) == IO(2)

>>> assert value_or(2)(Some(1)) == 1
>>> assert value_or(2)(Nothing) == 2

>>> assert anyio.run(
...      value_or(2)(ReaderFutureResult.from_value(1)),
...      ReaderFutureResult.empty,
... ) == IO(1)
>>> assert anyio.run(
...      value_or(2)(ReaderFutureResult.from_failure(1)),
...      ReaderFutureResult.empty,
... ) == IO(2)
← Converters Helper functions →