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 a useful program works. 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.
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.
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.
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!
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.
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
>>> from returns.pointfree import map_
>>> 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 = map_(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
>>> from returns.pointfree import map_
>>> 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 = map_(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
>>> from returns.pointfree import bind_result
>>> def regular_function(arg: int) -> Result[float, str]:
... if arg > 0:
... return Success(arg / 2)
... return Failure('zero')
>>> assert bind_result(regular_function)(
... IOSuccess(1),
... ) == IOResult.from_result(regular_function(1))
Lifting is useful when using returns.pipeline.pipe()
and other different declarative tools.
IOResult
values can be matched using the new feature of Python 3.10,
Structural Pattern Matching,
see the example below:
from returns.io import IOFailure, IOResult, IOSuccess
from returns.result import Success
container: IOResult[int, str] = IOSuccess(42)
match container:
# Matches if the result stored inside `IOSuccess` is `42`
# We need to use `Success` until the custom matching protocol
# is released. For more information, please visit:
# https://www.python.org/dev/peps/pep-0622/#custom-matching-protocol
case IOSuccess(Success(42)):
print('Result is "42"')
# Matches any `IOSuccess` instance
# and binds its value to the `value` variable
case IOSuccess(value):
print('Result is "{0}"'.format(value))
# Matches any `IOFailure` instance
case IOFailure(_):
print('A failure was occurred')
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.
Typing will only work correctly if our mypy plugin is used. This happens due to mypy issue.
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)
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.
Don’t forget to check out Converters.
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
>>> assert 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
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)
>>> assert lazy() == IO(1)
We have decided that it would be better and more familiar for Python devs.
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
.
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?
TLDR: you need to use IOSuccess
and IOFailure
functions
or IOResult.from_value
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_value(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.
Bases: returns.primitives.container.BaseContainer
, returns.primitives.hkt.SupportsKindN
[IO
, returns.io._ValueType
, NoReturn
, NoReturn
], returns.interfaces.specific.io.IOLikeN
[returns.io._ValueType
, NoReturn
, NoReturn
]
Explicit container for impure function results.
We also sometimes call it “marker” since once it is marked, it cannot be ever unmarked. 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.
See also
inner_value (+_ValueType) –
Typesafe equality comparison with other Result objects.
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')
function (Callable
[[+_ValueType], ~_NewValueType]) –
IO
[~_NewValueType]
Calls a wrapped function in a container on this container.
>>> from returns.io import IO
>>> assert IO('a').apply(IO(lambda inner: inner + 'b')) == IO('ab')
Or more complex example that shows how we can work
with regular functions and multiple IO
arguments:
>>> from returns.curry import curry
>>> @curry
... def appliable(first: str, second: str) -> str:
... return first + second
>>> assert IO('b').apply(IO('a').apply(IO(appliable))) == IO('ab')
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')
Alias for bind method. Part of the IOLikeN interface.
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.interfaces.applicative.ApplicativeN
interface.
inner_value (~_NewValueType) –
IO
[~_NewValueType]
Unit function to construct new IO
values from existing IO
.
>>> from returns.io import IO
>>> assert IO(1) == IO.from_io(IO(1))
Part of the returns.interfaces.specific.IO.IOLikeN
interface.
Converts IOResult[a, b]
back to IO[Result[a, b]]
.
Can be really helpful for composition.
>>> from returns.io import IO, IOSuccess
>>> from returns.result import Success
>>> assert IO.from_ioresult(IOSuccess(1)) == IO(Success(1))
Is the reverse of returns.io.IOResult.from_typecast()
.
Decorator to mark function that it returns IO
container.
If you need to mark async
function as impure,
use returns.future.future()
instead.
This decorator only works with sync functions. Example:
>>> from returns.io import IO, impure
>>> @impure
... def function(arg: int) -> int:
... return arg + 1 # this action is pure, just an example
...
>>> assert function(1) == IO(2)
function (Callable
[[~_FuncParams], ~_NewValueType]) –
Callable
[[~_FuncParams], IO
[~_NewValueType]]
Bases: returns.primitives.container.BaseContainer
, returns.primitives.hkt.SupportsKindN
[IOResult
, returns.io._ValueType
, returns.io._ErrorType
, NoReturn
], returns.interfaces.specific.ioresult.IOResultBasedN
[returns.io._ValueType
, returns.io._ErrorType
, NoReturn
]
Explicit container for impure function results that might fail.
Definition
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:
You can construct IOResult
from raw values
with IOSuccess()
and IOFailure()
public type constructors
You can construct IOResult
from IO
values
with from_failed_io()
and IOResult.from_io()
You can construct IOResult
from Result
values
with from_result()
We also have a lot of utility methods for better function composition like:
bind_result()
to work
with functions which return Result
from_typecast()
to work with IO[Result[...]]
values
Implementation
This class contains all the methods that can be delegated to Result
.
But, some methods are not implemented 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 functions and types instead.
inner_value (Result
[+_ValueType, +_ErrorType]) –
Success type that is used to represent the successful computation.
inner_value (+_ValueType) –
alias of returns.io.IOSuccess
Failure type that is used to represent the failed computation.
inner_value (+_ErrorType) –
alias of returns.io.IOFailure
Typesafe equality comparison with other IOResult objects.
Returns a stack trace when IOFailure()
was called.
Optional
[List
[FrameInfo
]]
Swaps value and error types.
So, values become errors and errors become values.
It is useful when you have to work with errors a lot.
And since we have a lot of .bind_
related methods
and only a single .lash
- it is easier to work with values.
>>> from returns.io import IOSuccess, IOFailure
>>> assert IOSuccess(1).swap() == IOFailure(1)
>>> assert IOFailure(1).swap() == IOSuccess(1)
IOResult
[+_ErrorType, +_ValueType]
Composes successful container with a pure function.
>>> from returns.io import IOSuccess
>>> assert IOSuccess(1).map(lambda num: num + 1) == IOSuccess(2)
function (Callable
[[+_ValueType], ~_NewValueType]) –
IOResult
[~_NewValueType, +_ErrorType]
Calls a wrapped function in a container on this container.
>>> from returns.io import IOSuccess, IOFailure
>>> def appliable(first: str) -> str:
... return first + 'b'
>>> assert IOSuccess('a').apply(
... IOSuccess(appliable),
... ) == IOSuccess('ab')
>>> assert IOFailure('a').apply(
... IOSuccess(appliable),
... ) == IOFailure('a')
>>> assert IOSuccess('a').apply(IOFailure(1)) == IOFailure(1)
>>> assert IOFailure('a').apply(IOFailure('b')) == IOFailure('a')
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')
Alias for bind_ioresult method. Part of the IOResultBasedN interface.
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')
Composes successful container with a function that returns a container.
Similar to bind()
, but works with containers
that return returns.io.IO
instead of IOResult
.
>>> from returns.io import IO, IOFailure, IOSuccess
>>> def bindable(string: str) -> IO[str]:
... return IO(string + 'z')
>>> assert IOSuccess('a').bind_io(bindable) == IOSuccess('az')
>>> assert IOFailure('a').bind_io(bindable) == IOFailure('a')
Composes failed container with a pure function to modify failure.
>>> from returns.io import IOFailure
>>> assert IOFailure(1).alt(float) == IOFailure(1.0)
function (Callable
[[+_ErrorType], ~_NewErrorType]) –
IOResult
[+_ValueType, ~_NewErrorType]
Composes failed container with a function that returns a container.
>>> from returns.io import IOFailure, IOSuccess, IOResult
>>> def lashable(state: str) -> IOResult[int, str]:
... if len(state) > 1:
... return IOSuccess(len(state))
... return IOFailure('oops')
>>> assert IOFailure('a').lash(lashable) == IOFailure('oops')
>>> assert IOFailure('abc').lash(lashable) == IOSuccess(3)
>>> assert IOSuccess('a').lash(lashable) == IOSuccess('a')
Get value from successful 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)
default_value (~_NewValueType) –
IO
[Union
[+_ValueType, ~_NewValueType]]
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
IO
[+_ValueType]
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
IO
[+_ErrorType]
Composes inner Result
with IOResult
returning function.
Can be useful when you need an access to both states of the result.
>>> from returns.io import IOResult, IOSuccess, IOFailure
>>> from returns.result import Result
>>> def count(container: Result[int, int]) -> IOResult[int, int]:
... return IOResult.from_result(
... container.map(lambda x: x + 1).alt(abs),
... )
>>> assert IOSuccess(1).compose_result(count) == IOSuccess(2)
>>> assert IOFailure(-1).compose_result(count) == IOFailure(1)
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)
Can be reverted via returns.io.IO.from_ioresult()
method.
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)
Creates new IOResult
from “successful” IO
container.
>>> from returns.io import IO, IOResult, IOSuccess
>>> container = IO(1)
>>> assert IOResult.from_io(container) == IOSuccess(1)
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)
Creates IOResult
from existing IOResult
value.
>>> from returns.io import IOResult, IOSuccess, IOFailure
>>> assert IOResult.from_ioresult(IOSuccess(1)) == IOSuccess(1)
>>> assert IOResult.from_ioresult(IOFailure(2)) == IOFailure(2)
One more value to create success unit values.
It is useful as a united way to create a new value from any container.
>>> from returns.io import IOResult, IOSuccess
>>> assert IOResult.from_value(1) == IOSuccess(1)
You can use this method or IOSuccess()
,
choose the most convenient for you.
inner_value (~_NewValueType) –
IOResult
[~_NewValueType, Any
]
One more value to create failure unit values.
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.
inner_value (~_NewErrorType) –
IOResult
[Any
, ~_NewErrorType]
Bases: returns.io.IOResult
[Any
, returns.io._ErrorType
]
IOFailure
representation.
inner_value (+_ErrorType) –
Alias for bind_ioresult method. Part of the IOResultBasedN interface. # noqa: E501
Bases: returns.io.IOResult
[returns.io._ValueType
, Any
]
IOSuccess
representation.
inner_value (+_ValueType) –
Alias for bind_ioresult method. Part of the IOResultBasedN interface. # noqa: E501
Alias for a popular case when IOResult
has Exception
as error type.
alias of returns.io.IOResult
[returns.io._ValueType
, Exception
]
Decorator to mark function that it returns IOResult
container.
Should be used with care, since it only catches Exception
subclasses.
It does not catch BaseException
subclasses.
If you need to mark async
function as impure,
use returns.future.future_safe()
instead.
This decorator only works with sync 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()
Similar to returns.future.future_safe()
and returns.result.safe()
decorators.
function (Callable
[[~_FuncParams], ~_NewValueType]) –
Callable
[[~_FuncParams], IOResult
[~_NewValueType, Exception
]]
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
>>> assert unsafe_perform_io(IO(1)) == 1
wrapped_in_io (IO
[~_ValueType]) –
~_ValueType