IO

Mathematicians dream in pure functions. Each of them only relies on its arguments and always produces the same result for the same input.

That’s not how useful program work. We need to rely on the environment and we need to do side effects.

Furthermore, there are several types of IO in our programs:

  • Some IO never fails, like: getting current date and time, random number, or OS name

  • Some IO might fail, like: sending network requests, accessing filesystem, or database

There’s a solution.

IO marker

Once you have an IO operation you can mark it appropriately. We can use a simple class returns.io.IO to mark impure parts of the program that do not fail.

>>> import random
>>> from returns.io import IO
>>> def get_random_number() -> IO[int]:
...     return IO(random.randint(1, 10))
...
>>> assert isinstance(get_random_number(), IO)

And later we can work inside this IO context and do not break into our pure part of the program:

>>> assert get_random_number().map(lambda number: number / number) == IO(1.0)

And it infects all other functions that call it.

>>> def modify_number(number: int) -> IO[float]:
...      return get_random_number().map(lambda rnd: number / rnd)
...
>>> assert isinstance(modify_number(1), IO)

It is good enough to indicate that you are aware of side effects of the function.

IOResult

On the other hand, we can have IO parts of the program that do fail.

Imagine we have this beautiful pure function:

def can_book_seats(
    number_of_seats: int,
    reservation: 'Reservation',
) -> bool:
    return reservation.capacity >= number_of_seats + reservation.booked

What’s good about it? We can test it easily. Even without setting up any testing framework, simple doctests will be enough.

This code is beautiful, because it is simple.

We can later use its result to process the result of the booking request:

def process_booking_result(is_successful: bool) -> 'ProcessID':
    ...

process_booking_result(is_successful)  # works just fine!

At this point we don’t have IO in our program.

Impure functions

But, imagine that our requirements had changed. And now we have to grab the number of already booked tickets from some other microservice and fetch the maximum capacity from the database:

import requests
import db

def can_book_seats(
    number_of_seats: int,
    place_id: int,
) -> bool:
    capacity = db.get_place_capacity(place_id)  # sql query
    booked = requests('https://partner.com/api').json()['booked']  # http req
    return capacity >= number_of_seats + booked

Now testing this code will become a nightmare! It will require to setup:

  • real database and tables

  • fixture data

  • requests mocks for different outcomes

  • and the whole Universe!

Our complexity has sky-rocketed! And the most annoying part is that all other functions that call can_book_seats now also have to do the same setup. It seams like IO is indelible mark (some people also call it “effect”).

And at some point it time we will start to mix pure and impure code together. Let’s not forget that all of these operations can fail too!

Separating two worlds

Well, IO mark is indeed indelible and should be respected.

And then impurity becomes explicit:

import requests
import db
from returns.io import IOResultE

def can_book_seats(
    number_of_seats: int,
    place_id: int,
) -> IOResultE[bool]:
    ...

Now this function returns IOResultE[bool] instead of a regular bool. It means, that it cannot be used where regular bool can be:

def process_booking_result(is_successful: bool) -> 'ProcessID':
    ...

is_successful: IOResultE[bool] = can_book_seats(number_of_seats, place_id)
process_booking_result(is_successful)  # Boom!
# => Argument 1 has incompatible type "IOResultE[bool]"; expected "bool"

See? It is now impossible for a pure function to use IOResultE[bool]. It is impossible to unwrap or get a raw value from this container. Once it is marked as IO it will never return to the pure state (well, there’s a hack actually: unsafe_perform_io).

Now we have to work inside the IO context:

message_id: IOResultE['ProcessID'] = can_book_seats(
    number_of_seats,
    place_id,
).map(
    process_booking_result,
)

Or it can be annotated to work with impure results:

def process_booking_result(
    is_successful: IOResultE[bool],
) -> IOResultE['ProcessID']:
    ...

is_successful: IOResult[bool] = can_book_seats(number_of_seats, place_id)
process_booking_result(is_successful)  # Works!

Now, all our impurity is explicit. We can track it, we can fight it, we can design it better. By saying that, it is assumed that you have a functional core and imperative shell.

Lifting

You can also lift regular functions into one that works with IO or IOResult on both ends. It really helps you with the composition!

>>> from returns.io import IO

>>> def regular_function(arg: int) -> float:
...     return arg / 2  # not an `IO` operation

>>> container = IO(1)
>>> # When we need to compose `regular_function` with `IO`,
>>> # we have two ways of doing it:
>>> io = container.map(regular_function)
>>> assert io == IO(0.5)

>>> # or, it is the same as:
>>> io = IO.lift(regular_function)(container)
>>> assert io == IO(0.5)

IOResult can lift both regular functions and ones that return Result:

>>> from returns.io import IOResult, IOSuccess

>>> def regular_function(arg: int) -> float:
...     return arg / 2  # not an `IO` operation

>>> container: IOResult[int, str] = IOSuccess(1)
>>> # When we need to compose `regular_function` with `IOResult`,
>>> # we have two ways of doing it:
>>> io = container.map(regular_function)
>>> assert io == IOSuccess(0.5)

>>> # or, it is the same as:
>>> io = IOResult.lift(regular_function)(container)
>>> assert io == IOSuccess(0.5)

And Result based functions:

>>> from returns.io import IOResult, IOSuccess
>>> from returns.result import Result, Success, Failure

>>> def regular_function(arg: int) -> Result[float, str]:
...     if arg > 0:
...         return Success(arg / 2)
...     return Failure('zero')
...

>>> assert IOResult.lift_result(regular_function)(
...     IOSuccess(1),
... ) == IOResult.from_result(regular_function(1))

Lifting is useful when using returns.pipeline.pipe() and other different declarative tools.

Aliases

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

  • returns.io.IOResultE is an alias for IOResult[... Exception], just use it when you want to work with IOResult containers that use exceptions as error type. It is named IOResultE because it is IOResultException and IOResultError at the same time.

Decorators

Limitations

Typing will only work correctly if decorator_plugin is used. This happens due to mypy issue.

impure

We also have this handy decorator to help you with the existing impure things in Python:

from returns.io import impure

name: IO[str] = impure(input)('What is your name?')

You can also decorate your own functions with @impure for better readability and clearness:

import random
from returns.io import impure

@impure
def get_user() -> 'User':
    return random.randint(1, 5)

impure_safe

Similar to impure and safe decorators. Once applied, it transforms the return type to be IOResultE:

from returns.io import IOResultE, impure_safe

@impure_safe
def http_get(path: str) -> 'Response':
    return requests.get(path)

container: IOResultE['Response'] = http_get('/home')

Use for impure operations that might fail.

Helpers

Don’t forget to check out Converters.

unsafe_perform_io

Sometimes you really need to get the raw value from IO container. For example:

def index_view(request, user_id):
    user: IO[User] = get_user(user_id)
    return render('index.html', {'user': user})  # ???

In this case your web-framework will not render your user correctly. Since it does not expect it to be wrapped inside IO containers. And we obviously cannot map or bind this function.

What to do? Use unsafe_perform_io:

from returns.unsafe import unsafe_perform_io

def index_view(request, user_id):
    user: IO[User] = get_user(user_id)
    return render('index.html', {'user': unsafe_perform_io(user)})  # Ok

We need it as an escape and compatibility mechanism for our imperative shell.

In other words:

>>> from returns.unsafe import unsafe_perform_io
>>> from returns.io import IO

>>> unsafe_perform_io(IO('abc'))
'abc'

It is recommended to use import-linter to restrict imports from returns.unsafe expect the top-level modules.

Inspired by Haskell’s unsafePerformIO

FAQ

Why aren’t IO lazy?

Please, note that our IO implementation is not lazy by design. This way when you mark something as @impure it will work as previously. The only thing that changes is the return type.

Instead we offer to use unsafe_perform_io to work with IO and simulate laziness.

But, you can always make your IO lazy:

>>> from returns.io import IO
>>> lazy = lambda: IO(1)
>>> str(lazy())
'<IO: 1>'

We have decided that it would be better and more familiar for Python devs.

What is the difference between IO[T] and T?

What kind of input parameter should my function accept IO[T] or simple T?

It really depends on your domain / context. If the value is pure, than use raw unwrapped values. If the value is fetched, input, received, selected, than use IO or IOResult container: first one for operations that never fail, second one for operations that might fail.

Most web applications are just fully covered with IO.

Why can’t we use IO[Result] instead of IOResult?

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

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

# Is the same as:

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

The second one looks better, doesn’t it?

How to create unit objects for IOResult?

TLDR: you need to use IOSuccess and IOFailure functions or IOResult.from_success and IOResult.from_failure methods:

>>> from returns.io import IOResult, IOSuccess, IOFailure
>>> first: IOResult[int, str] = IOSuccess(1)
>>> second: IOResult[float, int] = IOFailure(1)

>>> assert IOResult.from_success(1) == IOSuccess(1)
>>> assert IOResult.from_failure(2) == IOFailure(2)

You can also annotate your variables properly. Otherwise, mypy will treat IOSuccess(1) as IOSuccess[int, Any]. You can narrow the type in advance.

See How to create unit objects? for more details.

Why can’t we unwrap values or use @pipeline with IO?

Our design decision was not let people unwrap IO containers, so it will indeed infect the whole call-stack with its effect.

Otherwise, people might hack the system in some dirty (from our point of view) but valid (from the python’s point of view) ways.

Even IOResult can’t be unwrapped. When used together with @pipeline we will still receive IO values from returns.io.IOResult.unwrap() calls.

Warning:

Of course, you can directly access
the internal state of the IO with ``._internal_state``,
but your are considered to be a grown-up!

Use wemake-python-styleguide to restrict ._ access in your code.

API Reference

graph TD; _IOFailure IOResult _IOSuccess IO BaseContainer --> IO Generic --> IO BaseContainer --> IOResult Generic --> IOResult IOResult --> _IOFailure IOResult --> _IOSuccess
class IO(inner_value)[source]

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

Explicit marker for impure function results.

We call it “marker” since once it is marked, it cannot be unmarked.

IO is also a container. But, it is different in a way that it can’t be unwrapped / rescued / fixed. There’s no way to directly get its internal value.

Note that IO represents a computation that never fails.

Examples of such computations are:

  • read / write to localStorage

  • get the current time

  • write to the console

  • get a random number

Use IOResult[...] for operations that might fail. Like DB access or network operations.

map(function)[source]

Applies function to the inner value.

Applies ‘function’ to the contents of the IO instance and returns a new IO object containing the result. ‘function’ should accept a single “normal” (non-container) argument and return a non-container result.

>>> def mappable(string: str) -> str:
...      return string + 'b'
...
>>> assert IO('a').map(mappable) == IO('ab')
Return type

IO[~_NewValueType]

bind(function)[source]

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

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

>>> def bindable(string: str) -> IO[str]:
...      return IO(string + 'b')
...
>>> assert IO('a').bind(bindable) == IO('ab')
Return type

IO[~_NewValueType]

classmethod lift(function)[source]

Lifts function to be wrapped in IO for better composition.

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

Works similar to map(), but has inverse semantics.

This is how it should be used:

>>> from returns.io import IO
>>> def example(argument: int) -> float:
...     return argument / 2  # not exactly IO action!
...
>>> assert IO.lift(example)(IO(2)) == IO(1.0)
Return type

Callable[[IO[+_ValueType]], IO[~_NewValueType]]

classmethod from_value(inner_value)[source]

Unit function to construct new IO values.

Is the same as regular constructor:

>>> from returns.io import IO
>>> assert IO(1) == IO.from_value(1)

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

Return type

IO[~_NewValueType]

impure(function)[source]

Decorator to mark function that it returns IO container.

Supports both async and regular functions. Example:

>>> from returns.io import IO, impure
>>> @impure
... def function(arg: int) -> int:
...     return arg + 1
...
>>> assert function(1) == IO(2)
class IOResult(inner_value)[source]

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

Explicit marker for impure function results that might fail.

We call it “marker” since once it is marked, it cannot be unmarked.

This type is similar to returns.result.Result. This basically a more useful version of IO[Result[a, b]]. Use this type for IO computations that might fail. Examples of IO computations that might fail are:

  • access database

  • access network

  • access filesystem

Use IO for operations that do IO but do not fail.

Note, that even methods like unwrap`() and value_or() return values wrapped in IO.

IOResult is a complex compound value that consists of:

  • raw value

  • Result

  • IO

This is why it has so many helper and factory methods:

We also have a lot of utility methods for better function composition like:

This class contains all the methods that can be delegated to Result. But, some methods have raise NotImplementedError which means that we have to use special _IOSuccess and _IOFailure implementation details to correctly handle these callbacks.

Do not rely on them! Use public data.

success_type

alias of _IOSuccess

failure_type

alias of _IOFailure

map(function)[source]

Composes successful container with a pure function.

>>> from returns.io import IOSuccess
>>> assert IOSuccess(1).map(lambda num: num + 1) == IOSuccess(2)
Return type

IOResult[~_NewValueType, +_ErrorType]

bind(function)[source]

Composes successful container with a function that returns a container.

>>> from returns.io import IOResult, IOFailure, IOSuccess
>>> def bindable(string: str) -> IOResult[str, str]:
...      if len(string) > 1:
...          return IOSuccess(string + 'b')
...      return IOFailure(string + 'c')
...
>>> assert IOSuccess('aa').bind(bindable) == IOSuccess('aab')
>>> assert IOSuccess('a').bind(bindable) == IOFailure('ac')
>>> assert IOFailure('a').bind(bindable) == IOFailure('a')
Return type

IOResult[~_NewValueType, +_ErrorType]

bind_result(function)[source]

Composes successful container with a function that returns a container.

Similar to bind(), but works with containers that return returns.result.Result instead of IOResult.

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

>>> def bindable(string: str) -> Result[str, str]:
...      if len(string) > 1:
...          return Success(string + 'b')
...      return Failure(string + 'c')
...
>>> assert IOSuccess('aa').bind_result(bindable) == IOSuccess('aab')
>>> assert IOSuccess('a').bind_result(bindable) == IOFailure('ac')
>>> assert IOFailure('a').bind_result(bindable) == IOFailure('a')
Return type

IOResult[~_NewValueType, +_ErrorType]

fix(function)[source]

Composes failed container with a pure function to fix the failure.

>>> from returns.io import IOFailure, IOSuccess
>>> assert IOFailure('a').fix(
...     lambda char: char + 'b',
... ) == IOSuccess('ab')
Return type

IOResult[~_NewValueType, +_ErrorType]

alt(function)[source]

Composes failed container with a pure function to modify failure.

>>> from returns.io import IOFailure
>>> assert IOFailure(1).alt(float) == IOFailure(1.0)
Return type

IOResult[+_ValueType, ~_NewErrorType]

rescue(function)[source]

Composes failed container with a function that returns a container.

>>> from returns.io import IOFailure, IOSuccess, IOResult
>>> def rescuable(state: str) -> IOResult[int, str]:
...     if len(state) > 1:
...         return IOSuccess(len(state))
...     return IOFailure('oops')

>>> assert IOFailure('a').rescue(rescuable) == IOFailure('oops')
>>> assert IOFailure('abc').rescue(rescuable) == IOSuccess(3)
>>> assert IOSuccess('a').rescue(rescuable) == IOSuccess('a')
Return type

IOResult[+_ValueType, ~_NewErrorType]

value_or(default_value)[source]

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

>>> from returns.io import IO, IOFailure, IOSuccess
>>> assert IOSuccess(1).value_or(None) == IO(1)
>>> assert IOFailure(1).value_or(None) == IO(None)
Return type

IO[Union[+_ValueType, ~_NewValueType]]

unwrap()[source]

Get value from successful container or raise exception for failed one.

>>> from returns.io import IO, IOFailure, IOSuccess
>>> assert IOSuccess(1).unwrap() == IO(1)
>>> IOFailure(1).unwrap()
Traceback (most recent call last):
  ...
returns.primitives.exceptions.UnwrapFailedError
Return type

IO[+_ValueType]

failure()[source]

Get failed value from failed container or raise exception from success.

>>> from returns.io import IO, IOFailure, IOSuccess
>>> assert IOFailure(1).failure() == IO(1)
>>> IOSuccess(1).failure()
Traceback (most recent call last):
  ...
returns.primitives.exceptions.UnwrapFailedError
Return type

IO[+_ErrorType]

classmethod lift(function)[source]

Lifts function to be wrapped in IOResult for better composition.

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

Works similar to map(), but has inverse semantics.

This is how it should be used:

>>> from returns.io import IOResult, IOSuccess, IOFailure
>>> def example(argument: int) -> float:
...     return argument / 2  # not exactly IO action!
...
>>> assert IOResult.lift(example)(IOSuccess(2)) == IOSuccess(1.0)
>>> assert IOResult.lift(example)(IOFailure(2)) == IOFailure(2)

This one is similar to appling lift() and returns.result.Result.lift() in order.

Return type

Callable[[IOResult[+_ValueType, -_ContraErrorType]], IOResult[~_NewValueType, -_ContraErrorType]]

classmethod lift_result(function)[source]

Lifts function from Result to IOResult for better composition.

Similar to lift(), but works with other type.

>>> from returns.io import IOResult, IOSuccess
>>> from returns.result import Result, Success

>>> def returns_result(arg: int) -> Result[int, str]:
...     return Success(arg + 1)
...
>>> returns_ioresult = IOResult.lift_result(returns_result)
>>> assert returns_ioresult(IOSuccess(1)) == IOSuccess(2)
Return type

Callable[[IOResult[+_ValueType, +_ErrorType]], IOResult[~_NewValueType, +_ErrorType]]

classmethod from_typecast(container)[source]

Converts IO[Result[_ValueType, _ErrorType]] to IOResult.

Also prevails the type of Result to IOResult, so: IO[Result[_ValueType, _ErrorType]] would become IOResult[_ValueType, _ErrorType].

>>> from returns.result import Success
>>> from returns.io import IO, IOResult, IOSuccess
>>> container = IO(Success(1))
>>> assert IOResult.from_typecast(container) == IOSuccess(1)
Return type

IOResult[~_NewValueType, ~_NewErrorType]

classmethod from_failed_io(container)[source]

Creates new IOResult from “failed” IO container.

>>> from returns.io import IO, IOResult, IOFailure
>>> container = IO(1)
>>> assert IOResult.from_failed_io(container) == IOFailure(1)
Return type

IOResult[NoReturn, ~_NewErrorType]

classmethod from_successful_io(container)[source]

Creates new IOResult from “successful” IO container.

>>> from returns.io import IO, IOResult, IOSuccess
>>> container = IO(1)
>>> assert IOResult.from_successful_io(container) == IOSuccess(1)
Return type

IOResult[~_NewValueType, NoReturn]

classmethod from_result(container)[source]

Creates IOResult from Result value.

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

>>> assert IOResult.from_result(Success(1)) == IOSuccess(1)
>>> assert IOResult.from_result(Failure(2)) == IOFailure(2)
Return type

IOResult[~_NewValueType, ~_NewErrorType]

classmethod from_success(inner_value)[source]

One more value to create success unit values.

This is a part of returns.primitives.interfaces.Unitable. It is useful as a united way to create a new value from any container.

>>> from returns.io import IOResult, IOSuccess
>>> assert IOResult.from_success(1) == IOSuccess(1)

You can use this method or IOSuccess(), choose the most convenient for you.

Return type

IOResult[~_NewValueType, Any]

classmethod from_failure(inner_value)[source]

One more value to create failred unit values.

This is a part of returns.primitives.interfaces.Unitable. It is useful as a united way to create a new value from any container.

>>> from returns.io import IOResult, IOFailure
>>> assert IOResult.from_failure(1) == IOFailure(1)

You can use this method or IOFailure(), choose the most convenient for you.

Return type

IOResult[Any, ~_NewErrorType]

IOSuccess(inner_value)[source]

Public unit function of succeful IOResult container.

>>> from returns.io import IOSuccess
>>> str(IOSuccess(1))
'<IOResult: <Success: 1>>'
Return type

IOResult[~_NewValueType, Any]

IOFailure(inner_value)[source]

Public unit function of failed IOResult container.

>>> from returns.io import IOFailure
>>> str(IOFailure(1))
'<IOResult: <Failure: 1>>'
Return type

IOResult[Any, ~_NewErrorType]

IOResultE

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

alias of returns.io.IOResult

impure_safe(function)[source]

Decorator to mark function that it returns IO container.

Supports both async and regular functions. Example:

>>> from returns.io import IOSuccess, impure_safe
>>> @impure_safe
... def function(arg: int) -> float:
...     return 1 / arg
...
>>> assert function(1) == IOSuccess(1.0)
>>> assert function(0).failure()
unsafe_perform_io(wrapped_in_io)[source]

Compatibility utility and escape mechanism from IO world.

Just unwraps the internal value from returns.io.IO container. Should be used with caution! Since it might be overused by lazy and ignorant developers.

It is recommended to have only one place (module / file) in your program where you allow unsafe operations.

We recommend to use import-linter to enforce this rule:

>>> from returns.io import IO
>>> unsafe_perform_io(IO(1))
1
Return type

~_ValueType

Result Context