Source code for returns.io
# -*- coding: utf-8 -*-
from functools import wraps
from inspect import iscoroutinefunction
from typing import Callable, Coroutine, Generic, TypeVar, overload
from typing_extensions import final
from returns.primitives.container import BaseContainer
_ValueType = TypeVar('_ValueType')
_NewValueType = TypeVar('_NewValueType')
# Helpers:
_FirstType = TypeVar('_FirstType')
_SecondType = TypeVar('_SecondType')
[docs]@final
class IO(Generic[_ValueType], BaseContainer):
"""
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 cannot be unwrapped / rescued / fixed.
There's no way to directly get its internal value.
"""
_inner_value: _ValueType
def __init__(self, inner_value: _ValueType) -> None:
"""Required for typing."""
BaseContainer.__init__(self, inner_value) # type: ignore # noqa: Z462
[docs] def map( # noqa: A003
self,
function: Callable[[_ValueType], _NewValueType],
) -> 'IO[_NewValueType]':
"""
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.
"""
return IO(function(self._inner_value))
[docs] def bind(
self, function: Callable[[_ValueType], 'IO[_NewValueType]'],
) -> 'IO[_NewValueType]':
"""
Applies 'function' to the result of a previous calculation.
'function' should accept a single "normal" (non-container) argument
and return IO type object.
"""
return function(self._inner_value)
@overload
def impure( # type: ignore
function: Callable[..., Coroutine[_FirstType, _SecondType, _NewValueType]],
) -> Callable[
...,
Coroutine[_FirstType, _SecondType, IO[_NewValueType]],
]:
"""Case for async functions."""
@overload
def impure(
function: Callable[..., _NewValueType],
) -> Callable[..., IO[_NewValueType]]:
"""Case for regular functions."""
[docs]def impure(function):
"""
Decorator to mark function that it returns :py:class:`IO` container.
Supports both async and regular functions.
"""
if iscoroutinefunction(function):
async def decorator(*args, **kwargs):
return IO(await function(*args, **kwargs))
else:
def decorator(*args, **kwargs):
return IO(function(*args, **kwargs))
return wraps(function)(decorator)