PK!uhreturns/__init__.py# -*- coding: utf-8 -*- PK!uhreturns/contrib/__init__.py# -*- coding: utf-8 -*- PK!uh returns/contrib/mypy/__init__.py# -*- coding: utf-8 -*- PK!c4 (returns/contrib/mypy/decorator_plugin.py# -*- coding: utf-8 -*- """ Custom mypy plugin to solve the temporary problem with untyped decorators. This problem appears when we try to change the return type of the function. However, currently it is impossible due to this bug: https://github.com/python/mypy/issues/3157 This plugin is a temporary solution to the problem. It should be later replaced with the official way of doing things. ``mypy`` API docs are here: https://mypy.readthedocs.io/en/latest/extending_mypy.html We use ``pytest-mypy-plugins`` to test that it works correctly, see: https://github.com/mkurnikov/pytest-mypy-plugins """ from typing import Callable, Optional, Type from mypy.plugin import FunctionContext, Plugin from mypy.types import CallableType #: Set of full names of our decorators. _TYPED_DECORATORS = frozenset(( 'returns.result.safe', 'returns.io.impure', 'returns.maybe.maybe', )) def _change_decorator_function_type( decorator: CallableType, arg_type: CallableType, ) -> CallableType: """Replaces revealed argument types by mypy with types from decorated.""" return decorator.copy_modified( arg_types=arg_type.arg_types, arg_kinds=arg_type.arg_kinds, arg_names=arg_type.arg_names, variables=arg_type.variables, is_ellipsis_args=arg_type.is_ellipsis_args, ) def _analyze_decorator(function_ctx: FunctionContext): """Tells us what to do when one of the typed decorators is called.""" if not isinstance(function_ctx.arg_types[0][0], CallableType): return function_ctx.default_return_type if not isinstance(function_ctx.default_return_type, CallableType): return function_ctx.default_return_type return _change_decorator_function_type( function_ctx.default_return_type, function_ctx.arg_types[0][0], ) class _TypedDecoratorPlugin(Plugin): def get_function_hook( # type: ignore self, fullname: str, ) -> Optional[Callable[[FunctionContext], Type]]: """ One of the specified ``mypy`` callbacks. Runs on each function call in the source code. We are only interested in a particular subset of all functions. So, we return a function handler for them. Otherwise, we return ``None``. """ if fullname in _TYPED_DECORATORS: return _analyze_decorator return None def plugin(version: str) -> Type[Plugin]: """Plugin's public API and entrypoint.""" return _TypedDecoratorPlugin PK!-?ereturns/converters.py# -*- coding: utf-8 -*- from typing import Callable, TypeVar, overload from returns.io import IO from returns.maybe import Maybe from returns.pipeline import is_successful from returns.result import Failure, Result, Success # Contianer internals: _ValueType = TypeVar('_ValueType') _ErrorType = TypeVar('_ErrorType') # Aliases: _FirstType = TypeVar('_FirstType') def result_to_maybe( result_container: Result[_ValueType, _ErrorType], ) -> Maybe[_ValueType]: """ Converts ``Result`` container to ``Maybe`` container. .. code:: python >>> from returns.maybe import Some, Nothing >>> from returns.result import Failure, Success >>> result_to_maybe(Success(1)) == Some(1) True >>> result_to_maybe(Failure(1)) == Nothing True >>> result_to_maybe(Success(None)) == Nothing True """ return Maybe.new(result_container.value_or(None)) def maybe_to_result( maybe_container: Maybe[_ValueType], ) -> Result[_ValueType, None]: """ Converts ``Maybe`` container to ``Result`` container. .. code:: python >>> from returns.maybe import Some, Nothing >>> from returns.result import Failure, Success >>> maybe_to_result(Nothing) == Failure(None) True >>> maybe_to_result(Some(1)) == Success(1) True >>> maybe_to_result(Some(None)) == Failure(None) True """ inner_value = maybe_container.value_or(None) if inner_value is not None: return Success(inner_value) return Failure(inner_value) @overload def join(container: IO[IO[_ValueType]]) -> IO[_ValueType]: """Case for ``IO`` container.""" @overload def join(container: Maybe[Maybe[_ValueType]]) -> Maybe[_ValueType]: """Case for ``Maybe`` container.""" @overload def join( container: Result[Result[_ValueType, _ErrorType], _ErrorType], ) -> Result[_ValueType, _ErrorType]: """Case for ``Result`` container.""" def join(container): """ Joins two nested containers together. .. code:: python >>> from returns.maybe import Some >>> from returns.result import Success >>> from returns.io import IO >>> join(IO(IO(1))) == IO(1) True >>> join(Some(Some(1))) == Some(1) True >>> join(Success(Success(1))) == Success(1) True """ return container._inner_value # noqa: WPS437 _coalesce_doc = """ Accepts two functions that handle different cases of containers. First one handles successful containers like ``Some`` and ``Success``, and second one for failed containers like ``Nothing`` and ``Failure``. This function is useful when you need to coalesce two possible container states into one type. """ def _coalesce(success_handler, failure_handler): """ We need this function, because we cannot use a single typed function. .. code:: python >>> from returns.result import Success, Failure >>> f1 = lambda x: x + 1 >>> f2 = lambda y: y + 'b' >>> coalesce_result(f1, f2)(Success(1)) == 2 True >>> coalesce_result(f1, f2)(Failure('a')) == 'ab' True >>> from returns.maybe import Some, Nothing >>> f1 = lambda x: x + 1 >>> f2 = lambda _: 'a' >>> coalesce_maybe(f1, f2)(Some(1)) == 2 True >>> coalesce_maybe(f1, f2)(Nothing) == 'a' True """ def decorator(container): if is_successful(container): return success_handler(container.unwrap()) return failure_handler(container.failure()) return decorator coalesce_result: Callable[ [ Callable[[_ValueType], _FirstType], Callable[[_ErrorType], _FirstType], ], Callable[[Result[_ValueType, _ErrorType]], _FirstType], ] = _coalesce coalesce_result.__doc__ = _coalesce_doc coalesce_maybe: Callable[ [ Callable[[_ValueType], _FirstType], Callable[[None], _FirstType], ], Callable[[Maybe[_ValueType]], _FirstType], ] = _coalesce coalesce_maybe.__doc__ = _coalesce_doc PK!t2 2 returns/functions.py# -*- coding: utf-8 -*- from typing import Any, Callable, NoReturn, TypeVar from returns.generated.box import _box as box # noqa: F401, WPS436 # Aliases: _FirstType = TypeVar('_FirstType') _SecondType = TypeVar('_SecondType') _ThirdType = TypeVar('_ThirdType') def identity(instance: _FirstType) -> _FirstType: """ Function that returns its argument. .. code:: python >>> identity(1) 1 >>> identity([1, 2, 3]) [1, 2, 3] Why do we even need this? Identity functions help us with the composition. Imagine, that you want to use :func:`returns.converters.coalesce_result` like so: .. code:: python from returns.result import Result from returns.converters import coalesce_result numbers: Result[int, float] # Now you want to fold `number` into `int` type: number: int = coalesce_result(identity, int)(numbers) # Done! See also: - https://en.wikipedia.org/wiki/Identity_function - https://stackoverflow.com/a/21506571/4842742 """ return instance def compose( first: Callable[[_FirstType], _SecondType], second: Callable[[_SecondType], _ThirdType], ) -> Callable[[_FirstType], _ThirdType]: """ Allows function composition. Works as: ``second . first`` or ``first() |> second()``. You can read it as "second after first". .. code:: python >>> compose(float, int)('123.5') 123 We can only compose functions with one argument and one return. Type checked. """ return lambda argument: second(first(argument)) def tap( function: Callable[[_FirstType], Any], ) -> Callable[[_FirstType], _FirstType]: """ Allows to apply some function and return an argument, instead of a result. Is usefull for side-effects like ``print()``, ``logger.log``, etc. .. code:: python >>> tap(print)(1) 1 1 >>> tap(lambda _: 1)(2) 2 See also: - https://github.com/dry-python/returns/issues/145 """ def decorator(argument_to_return: _FirstType) -> _FirstType: function(argument_to_return) return argument_to_return return decorator def raise_exception(exception: Exception) -> NoReturn: """ Helper function to raise exceptions as a function. It might be required as a compatibility tool for existing APIs. That's how it can be used: .. code:: python >>> from returns.result import Failure, Result >>> # Some operation result: >>> user: Result[int, ValueError] = Failure(ValueError('boom')) >>> # Here we unwrap internal exception and raise it: >>> user.fix(raise_exception) Traceback (most recent call last): ... ValueError: boom See also: - https://github.com/dry-python/returns/issues/56 """ raise exception PK!dreturns/generated/__init__.py# -*- coding: utf-8 -*- """ This package contains code that was "generated" via metaprogramming. This happens, because python is not flexible enough to do most common tasks in typed functional programming. Policy: 1. We store implementations in regular ``.py`` files. 2. We store generated type annotations in ``.pyi`` files. 3. We re-export these functions into regular modules as public values. Please, do not touch this code unless you know what you are doing. """ PK! ooreturns/generated/box.py# -*- coding: utf-8 -*- def _box(function): """ Boxes 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]`` This is how it should be used: .. code:: python >>> from returns.functions import box >>> from returns.maybe import Maybe, Some >>> def example(argument: int) -> Maybe[int]: ... return Some(argument + 1) ... >>> box(example)(Some(1)) == Some(2) True """ return lambda container: container.bind(function) PK!returns/generated/box.pyi# -*- coding: utf-8 -*- from typing import Callable, TypeVar, overload from returns.io import IO from returns.maybe import Maybe from returns.result import Result _ValueType = TypeVar('_ValueType') _ErrorType = TypeVar('_ErrorType') _NewValueType = TypeVar('_NewValueType') # Box: @overload def _box( function: Callable[[_ValueType], Maybe[_NewValueType]], ) -> Callable[[Maybe[_ValueType]], Maybe[_NewValueType]]: ... @overload def _box( function: Callable[[_ValueType], IO[_NewValueType]], ) -> Callable[[IO[_ValueType]], IO[_NewValueType]]: ... @overload def _box( function: Callable[[_ValueType], Result[_NewValueType, _ErrorType]], ) -> Callable[ [Result[_ValueType, _ErrorType]], Result[_NewValueType, _ErrorType], ]: ... PK!returns/generated/pipe.py# -*- coding: utf-8 -*- from functools import reduce from returns.functions import compose def _pipe(*functions): """ Allows to compose a value and up to 7 functions that use this value. Each next function uses the previos result as an input parameter. Here's how it should be used: .. code:: python >>> from returns.pipeline import pipe # => executes: str(float(int('1'))) >>> pipe(int, float, str)('1') '1.0' See also: - https://stackoverflow.com/a/41585450/4842742 - https://github.com/gcanti/fp-ts/blob/master/src/pipeable.ts """ return lambda initial: reduce(compose, functions)(initial) PK!6Mreturns/generated/pipe.pyi# -*- coding: utf-8 -*- from typing import Callable, TypeVar, overload _T1 = TypeVar('_T1') _T2 = TypeVar('_T2') _T3 = TypeVar('_T3') _T4 = TypeVar('_T4') _T5 = TypeVar('_T5') _T6 = TypeVar('_T6') _T7 = TypeVar('_T7') _T8 = TypeVar('_T8') _T9 = TypeVar('_T9') @overload def _pipe( p1: Callable[[_T1], _T2], p2: Callable[[_T2], _T3], ) -> Callable[[_T1], _T3]: ... @overload def _pipe( p1: Callable[[_T1], _T2], p2: Callable[[_T2], _T3], p3: Callable[[_T3], _T4], ) -> Callable[[_T1], _T4]: ... @overload def _pipe( p1: Callable[[_T1], _T2], p2: Callable[[_T2], _T3], p3: Callable[[_T3], _T4], p4: Callable[[_T4], _T5], ) -> Callable[[_T1], _T5]: ... @overload # noqa: WPS211 def _pipe( p1: Callable[[_T1], _T2], p2: Callable[[_T2], _T3], p3: Callable[[_T3], _T4], p4: Callable[[_T4], _T5], p5: Callable[[_T5], _T6], ) -> Callable[[_T1], _T6]: ... @overload # noqa: WPS211 def _pipe( p1: Callable[[_T1], _T2], p2: Callable[[_T2], _T3], p3: Callable[[_T3], _T4], p4: Callable[[_T4], _T5], p5: Callable[[_T5], _T6], p6: Callable[[_T6], _T7], ) -> Callable[[_T1], _T7]: ... @overload # noqa: WPS211 def _pipe( p1: Callable[[_T1], _T2], p2: Callable[[_T2], _T3], p3: Callable[[_T3], _T4], p4: Callable[[_T4], _T5], p5: Callable[[_T5], _T6], p6: Callable[[_T6], _T7], p7: Callable[[_T7], _T8], ) -> Callable[[_T1], _T8]: ... @overload # noqa: WPS211 def _pipe( p1: Callable[[_T1], _T2], p2: Callable[[_T2], _T3], p3: Callable[[_T3], _T4], p4: Callable[[_T4], _T5], p5: Callable[[_T5], _T6], p6: Callable[[_T6], _T7], p7: Callable[[_T7], _T8], p8: Callable[[_T8], _T9], ) -> Callable[[_T1], _T9]: ... PK!h xx returns/io.py# -*- 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', covariant=True) _NewValueType = TypeVar('_NewValueType') # Helpers: _FirstType = TypeVar('_FirstType') _SecondType = TypeVar('_SecondType') @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) # noqa: WPS609 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. .. code:: python >>> def mappable(string: str) -> str: ... return string + 'b' ... >>> IO('a').map(mappable) == IO('ab') True """ return IO(function(self._inner_value)) 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. .. code:: python >>> def bindable(string: str) -> IO[str]: ... return IO(string + 'b') ... >>> IO('a').bind(bindable) == IO('ab') True """ return function(self._inner_value) @classmethod def lift( cls, function: Callable[[_ValueType], _NewValueType], ) -> Callable[['IO[_ValueType]'], 'IO[_NewValueType]']: """ Lifts function to be wrapped in ``IO`` for better composition. In other words, it modifies the function signature from: ``a -> b`` to: ``IO[a] -> IO[b]`` This is how it should be used: .. code:: python >>> def example(argument: int) -> float: ... return argument / 2 # not exactly IO action! ... >>> IO.lift(example)(IO(2)) == IO(1.0) True See also: - https://wiki.haskell.org/Lifting - https://github.com/witchcrafters/witchcraft - https://en.wikipedia.org/wiki/Natural_transformation """ return lambda container: cls.map(container, function) @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.""" 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): # noqa: WPS430 return IO(await function(*args, **kwargs)) else: def decorator(*args, **kwargs): # noqa: WPS430 return IO(function(*args, **kwargs)) return wraps(function)(decorator) PK!vf((((returns/maybe.py# -*- coding: utf-8 -*- from abc import ABCMeta from functools import wraps from inspect import iscoroutinefunction from typing import ( Any, Callable, Coroutine, Generic, NoReturn, Optional, TypeVar, Union, overload, ) from typing_extensions import final from returns.primitives.container import BaseContainer from returns.primitives.exceptions import UnwrapFailedError # Definitions: _ValueType = TypeVar('_ValueType', covariant=True) _NewValueType = TypeVar('_NewValueType') # Aliases: _FirstType = TypeVar('_FirstType') _SecondType = TypeVar('_SecondType') class Maybe( Generic[_ValueType], BaseContainer, metaclass=ABCMeta, ): """ Represents a result of a series of commutation that can return ``None``. An alternative to using exceptions or constant ``is None`` checks. ``Maybe`` is an abstract type and should not be instantiated directly. Instead use ``Some`` and ``Nothing``. """ _inner_value: Optional[_ValueType] @classmethod def new(cls, inner_value: Optional[_ValueType]) -> 'Maybe[_ValueType]': """Creates new instance of Maybe container based on a value.""" if inner_value is None: return _Nothing(inner_value) return _Some(inner_value) def map( # noqa: A003 self, function: Callable[[_ValueType], Optional[_NewValueType]], ) -> 'Maybe[_NewValueType]': """Abstract method to compose container with a pure function.""" raise NotImplementedError def bind( self, function: Callable[[_ValueType], 'Maybe[_NewValueType]'], ) -> 'Maybe[_NewValueType]': """Abstract method to compose container with other container.""" raise NotImplementedError def fix( self, function: Callable[[None], Optional[_NewValueType]], ) -> 'Maybe[_NewValueType]': """Abstract method to compose container with a pure function.""" raise NotImplementedError def rescue( self, function: Callable[[None], 'Maybe[_NewValueType]'], ) -> 'Maybe[_NewValueType]': """Abstract method to compose container with other container.""" raise NotImplementedError def value_or( self, default_value: _NewValueType, ) -> Union[_ValueType, _NewValueType]: """Get value or default value.""" raise NotImplementedError def unwrap(self) -> _ValueType: """Get value or raise exception.""" raise NotImplementedError def failure(self) -> None: """Get failed value or raise exception.""" raise NotImplementedError @final class _Nothing(Maybe[Any]): """Represents an empty state.""" _inner_value: None def __init__(self, inner_value: None = None) -> None: """ Wraps the given value in the Container. ``inner_value`` can only be ``None``. """ BaseContainer.__init__(self, inner_value) # noqa: WPS609 def __str__(self): """Custom str definition without state inside.""" return '' def map(self, function): # noqa: A003 """ Returns the 'Nothing' instance that was used to call the method. .. code:: python >>> def mappable(string: str) -> str: ... return string + 'b' ... >>> Nothing.map(mappable) == Nothing True """ return self def bind(self, function): """ Returns the 'Nothing' instance that was used to call the method. .. code:: python >>> def bindable(string: str) -> Maybe[str]: ... return Some(string + 'b') ... >>> Nothing.bind(bindable) == Nothing True """ return self def fix(self, function): """ Applies function to the inner value. Applies 'function' to the contents of the 'Some' instance and returns a new 'Some' object containing the result. 'function' should not accept any arguments and return a non-container result. .. code:: python >>> def fixable(_state) -> str: ... return 'ab' ... >>> Nothing.fix(fixable) == Some('ab') True """ return Maybe.new(function(self._inner_value)) def rescue(self, function): """ Applies 'function' to the result of a previous calculation. 'function' should not accept any arguments and return Maybe a 'Nothing' or 'Some' type object. .. code:: python >>> def rescuable(_state) -> Maybe[str]: ... return Some('ab') ... >>> Nothing.rescue(rescuable) == Some('ab') True """ return function(self._inner_value) def value_or(self, default_value): """ Returns the value if we deal with 'Some' or default if 'Nothing'. .. code:: python >>> Nothing.value_or(1) 1 """ return default_value def unwrap(self): """ Raises an exception, since it does not have a value inside. .. code:: python >>> Nothing.unwrap() Traceback (most recent call last): ... returns.primitives.exceptions.UnwrapFailedError """ raise UnwrapFailedError(self) def failure(self) -> None: """ Get failed value. .. code:: python >>> Nothing.failure() is None True """ return self._inner_value @final class _Some(Maybe[_ValueType]): """ Represents a calculation which has succeeded and contains the value. Quite similar to ``Success`` type. """ _inner_value: _ValueType def __init__(self, inner_value: _ValueType) -> None: """Required for typing.""" BaseContainer.__init__(self, inner_value) # noqa: WPS609 def map(self, function): # noqa: A003 """ Applies function to the inner value. Applies 'function' to the contents of the 'Some' instance and returns a new 'Maybe' object containing the result. 'function' should accept a single "normal" (non-container) argument and return a non-container result. .. code:: python >>> def mappable(string: str) -> str: ... return string + 'b' ... >>> Some('a').map(mappable) == Some('ab') True """ return Maybe.new(function(self._inner_value)) def bind(self, function): """ Applies 'function' to the result of a previous calculation. 'function' should accept a single "normal" (non-container) argument and return 'Nothing' or 'Some' type object. .. code:: python >>> def bindable(string: str) -> Maybe[str]: ... return Some(string + 'b') ... >>> Some('a').bind(bindable) == Some('ab') True """ return function(self._inner_value) def fix(self, function): """ Returns the 'Some' instance that was used to call the method. .. code:: python >>> def fixable(_state) -> str: ... return 'ab' ... >>> Some('a').fix(fixable) == Some('a') True """ return self def rescue(self, function): """ Returns the 'Some' instance that was used to call the method. .. code:: python >>> def rescuable(_state) -> Maybe[str]: ... return Some('ab') ... >>> Some('a').rescue(rescuable) == Some('a') True """ return self def value_or(self, default_value): """ Returns the value if we deal with 'Some' or default if 'Nothing'. .. code:: python >>> Some(1).value_or(2) 1 """ return self._inner_value def unwrap(self): """ Returns the unwrapped value from the inside of this container. .. code:: python >>> Some(1).unwrap() 1 """ return self._inner_value def failure(self): """ Raises an exception, since it does not have a failure inside. .. code:: python >>> Some(1).failure() Traceback (most recent call last): ... returns.primitives.exceptions.UnwrapFailedError """ raise UnwrapFailedError(self) def Some(inner_value: Optional[_ValueType]) -> Maybe[_ValueType]: # noqa: N802 """Public unit function of protected `_Some` type.""" return Maybe.new(inner_value) #: Public unit value of protected `_Nothing` type. Nothing: Maybe[NoReturn] = _Nothing() @overload def maybe( # type: ignore function: Callable[ ..., Coroutine[_FirstType, _SecondType, Optional[_ValueType]], ], ) -> Callable[ ..., Coroutine[_FirstType, _SecondType, Maybe[_ValueType]], ]: """Case for async functions.""" @overload def maybe( function: Callable[..., Optional[_ValueType]], ) -> Callable[..., Maybe[_ValueType]]: """Case for regular functions.""" def maybe(function): """ Decorator to covert ``None`` returning function to ``Maybe`` container. Supports both async and regular functions. .. code:: python >>> from typing import Optional >>> @maybe ... def might_be_none(arg: int) -> Optional[int]: ... if arg == 0: ... return None ... return 1 / arg ... >>> might_be_none(0) == Nothing True >>> might_be_none(1) == Some(1.0) True """ if iscoroutinefunction(function): async def decorator(*args, **kwargs): # noqa: WPS430 regular_result = await function(*args, **kwargs) if regular_result is None: return Nothing return Some(regular_result) else: def decorator(*args, **kwargs): # noqa: WPS430 regular_result = function(*args, **kwargs) if regular_result is None: return Nothing return Some(regular_result) return wraps(function)(decorator) PK!f f returns/pipeline.py# -*- coding: utf-8 -*- from functools import wraps from inspect import iscoroutinefunction from typing import Callable, Coroutine, TypeVar, Union, overload from returns.generated.pipe import _pipe as pipe # noqa: F401, WPS436 from returns.maybe import Maybe from returns.primitives.exceptions import UnwrapFailedError from returns.result import Result # Logical aliases: _Unwrapable = Union[Result, Maybe] # Just aliases: _FirstType = TypeVar('_FirstType') _SecondType = TypeVar('_SecondType') # Hacks for functions: _ReturnsResultType = TypeVar( '_ReturnsResultType', bound=Callable[..., _Unwrapable], ) _AsyncReturnsResultType = TypeVar( '_AsyncReturnsResultType', bound=Callable[..., Coroutine[_FirstType, _SecondType, _Unwrapable]], ) def is_successful(container: _Unwrapable) -> bool: """ Determins if a container was successful or not. We treat container that raise ``UnwrapFailedError`` on ``.unwrap()`` not successful. .. code:: python >>> from returns.maybe import Some, Nothing >>> from returns.result import Failure, Success >>> is_successful(Some(1)) True >>> is_successful(Nothing) False >>> is_successful(Success(1)) True >>> is_successful(Failure(1)) False """ try: container.unwrap() except UnwrapFailedError: return False else: return True @overload def pipeline( function: _AsyncReturnsResultType, ) -> _AsyncReturnsResultType: """Case for async functions.""" @overload def pipeline(function: _ReturnsResultType) -> _ReturnsResultType: """Case for regular functions.""" def pipeline(function): # noqa: C901 """ Decorator to enable ``do-notation`` context. Should be used for series of computations that rely on ``.unwrap`` method. Supports both async and regular functions. """ if iscoroutinefunction(function): async def decorator(*args, **kwargs): # noqa: WPS430 try: return await function(*args, **kwargs) except UnwrapFailedError as exc: return exc.halted_container else: def decorator(*args, **kwargs): # noqa: WPS430 try: return function(*args, **kwargs) except UnwrapFailedError as exc: return exc.halted_container return wraps(function)(decorator) PK!uhreturns/primitives/__init__.py# -*- coding: utf-8 -*- PK!۬returns/primitives/container.py# -*- coding: utf-8 -*- from abc import ABCMeta from typing import Any, Callable, NoReturn, TypeVar, Union from typing_extensions import Protocol, runtime from returns.primitives.exceptions import ImmutableStateError _ValueType = TypeVar('_ValueType', covariant=True) _NewValueType = TypeVar('_NewValueType') _ErrorType = TypeVar('_ErrorType', covariant=True) _NewErrorType = TypeVar('_NewErrorType') class BaseContainer(object, metaclass=ABCMeta): """Utility class to provide all needed magic methods to the context.""" __slots__ = ('_inner_value',) _inner_value: Any def __init__(self, inner_value) -> None: """ Wraps the given value in the Container. 'value' is any arbitrary value of any type including functions. """ object.__setattr__(self, '_inner_value', inner_value) # noqa: WPS609 def __setattr__(self, attr_name: str, attr_value) -> NoReturn: """Makes inner state of the containers immutable.""" raise ImmutableStateError() def __delattr__(self, attr_name: str) -> NoReturn: # noqa: WPS603 """Makes inner state of the containers immutable.""" raise ImmutableStateError() def __str__(self) -> str: """Converts to string.""" return '<{0}: {1}>'.format( self.__class__.__qualname__.strip('_'), str(self._inner_value), ) def __eq__(self, other) -> bool: """Used to compare two 'Container' objects.""" if not isinstance(self, type(other)): return False return self._inner_value == other._inner_value # noqa: WPS437 def __hash__(self) -> int: """Used to use this value as a key.""" return hash(self._inner_value) @runtime class Bindable(Protocol[_ValueType]): """ Represents a "context" in which calculations can be executed. ``Bindable`` allows you to bind together a series of calculations while maintaining the context of that specific container. """ def bind( self, function: Callable[[_ValueType], 'Bindable[_NewValueType]'], ) -> 'Bindable[_NewValueType]': """ Applies 'function' to the result of a previous calculation. And returns a new container. Works for containers that represent success. Is the opposite of :meth:`Rescueable.rescue`. """ @runtime class Mappable(Protocol[_ValueType]): """ Allows to chain wrapped values with regular functions. Behaves like functor. """ def map( # noqa: A003 self, function: Callable[[_ValueType], _NewValueType], ) -> 'Mappable[_NewValueType]': """ Applies 'function' to the contents of the functor. And returns a new functor value. Is the opposite of :meth:`Fixable.fix`. """ @runtime class Fixable(Protocol[_ValueType, _ErrorType]): """Represents containers that can be fixed and rescued.""" def fix( self, function: Callable[[_ErrorType], _NewValueType], ) -> 'Fixable[_NewValueType, _ErrorType]': """ Applies 'function' to the error and transforms failure to success. And returns a new functor value. Works for containers that represent failure. Is the opposite of :meth:`Mappable.map`. """ @runtime class Rescueable(Protocol[_NewValueType, _ErrorType]): """ Represents a "context" in which calculations can be executed. ``Rescueable`` allows you to bind together a series of calculations while maintaining the context of that specific container. """ def rescue( self, function: Callable[ [_ErrorType], 'Rescueable[_NewValueType, _NewErrorType]', ], ) -> 'Rescueable[_NewValueType, _NewErrorType]': """ Applies 'function' to the result of a previous calculation. And returns a new container. Works for containers that represent failure. Is the opposite of :meth:`~bind`. """ @runtime class Unwrapable(Protocol[_ValueType]): """Represents containers that can unwrap and return its wrapped value.""" def value_or( self, default_value: _NewValueType, ) -> Union[_ValueType, _NewValueType]: """Forces to unwrap value from container or return a default.""" def unwrap(self) -> _ValueType: """ Custom magic method to unwrap inner value from container. Should be redefined for ones that actually have values. And for ones that raise an exception for no values. This method is the opposite of :meth:`~failure`. """ def failure(self) -> _ErrorType: """ Custom magic method to unwrap inner value from the failed container. This method is the opposite of :meth:`~unwrap`. """ @runtime class UnwrapableFailure(Protocol[_ValueType, _ErrorType]): """Allows to unwrap failures.""" def alt( self, function: Callable[[_ErrorType], _NewErrorType], ) -> 'Fixable[_ValueType, _NewErrorType]': """ Uses 'function' to transform one error to another. And returns a new functor value. Works for containers that represent failure. Is the opposite of :meth:`~map`. """ PK! None: """ Saves halted container in the inner state. So, this container can later be unpacked from this exception and used as a regular value. """ super().__init__() self.halted_container = container class ImmutableStateError(Exception): """Raised when a container is forced to be mutated.""" PK!returns/py.typedPK!2 / /returns/result.py# -*- coding: utf-8 -*- from abc import ABCMeta from functools import wraps from inspect import iscoroutinefunction from typing import ( Any, Callable, Coroutine, Generic, NoReturn, TypeVar, Union, overload, ) from typing_extensions import final from returns.primitives.container import BaseContainer from returns.primitives.exceptions import UnwrapFailedError # Definitions: _ValueType = TypeVar('_ValueType', covariant=True) _NewValueType = TypeVar('_NewValueType') _ErrorType = TypeVar('_ErrorType', covariant=True) _NewErrorType = TypeVar('_NewErrorType') # Aliases: _DefaultValueType = TypeVar('_DefaultValueType') _FirstType = TypeVar('_FirstType') _SecondType = TypeVar('_SecondType') class Result( Generic[_ValueType, _ErrorType], BaseContainer, metaclass=ABCMeta, ): """ Base class for :class:`~_Failure` and :class:`~_Success`. :class:`~Result` does not have """ _inner_value: Union[_ValueType, _ErrorType] def map( # noqa: A003 self, function: Callable[[_ValueType], _NewValueType], ) -> 'Result[_NewValueType, _ErrorType]': """Abstract method to compose container with a pure function.""" raise NotImplementedError def bind( self, function: Callable[[_ValueType], 'Result[_NewValueType, _ErrorType]'], ) -> 'Result[_NewValueType, _ErrorType]': """Abstract method to compose a container with another container.""" raise NotImplementedError def fix( self, function: Callable[[_ErrorType], _NewValueType], ) -> 'Result[_NewValueType, _ErrorType]': """ Abstract method to compose failed container and a pure function. This pure function should return a new state for a successful container. """ raise NotImplementedError def alt( self, function: Callable[[_ErrorType], _NewErrorType], ) -> 'Result[_ValueType, _NewErrorType]': """ Abstract method to compose failed container and a pure function. This pure function should return a new state for a new failed container. """ raise NotImplementedError def rescue( self, function: Callable[ [_ErrorType], 'Result[_ValueType, _NewErrorType]', ], ) -> 'Result[_ValueType, _NewErrorType]': """ Abstract method to compose a failed container with another container. This method is the oposite of ``.bind()``. """ raise NotImplementedError def value_or( self, default_value: _DefaultValueType, ) -> Union[_ValueType, _DefaultValueType]: """Get value or default value.""" raise NotImplementedError def unwrap(self) -> _ValueType: """Get value or raise exception.""" raise NotImplementedError def failure(self) -> _ErrorType: """Get failed value or raise exception.""" raise NotImplementedError @final class _Failure(Result[Any, _ErrorType]): """ Represents a calculation which has failed. It should contain an error code or message. Should not be used directly. """ _inner_value: _ErrorType def __init__(self, inner_value: _ErrorType) -> None: """Required for typing.""" BaseContainer.__init__(self, inner_value) # noqa: WPS609 def map(self, function): # noqa: A003 """ Returns the '_Failure' instance that was used to call the method. .. code:: python >>> def mappable(string: str) -> str: ... return string + 'b' ... >>> Failure('a').map(mappable) == Failure('a') True """ return self def bind(self, function): """ Returns the '_Failure' instance that was used to call the method. .. code:: python >>> def bindable(string: str) -> Result[str, str]: ... return Success(string + 'b') ... >>> Failure('a').bind(bindable) == Failure('a') True """ return self def fix(self, function): """ Applies function to the inner value. Applies 'function' to the contents of the '_Success' instance and returns a new '_Success' object containing the result. 'function' should accept a single "normal" (non-container) argument and return a non-container result. .. code:: python >>> def fixable(arg: str) -> str: ... return 'ab' ... >>> Failure('a').fix(fixable) == Success('ab') True """ return _Success(function(self._inner_value)) def rescue(self, function): """ Applies 'function' to the result of a previous calculation. 'function' should accept a single "normal" (non-container) argument and return Result a '_Failure' or '_Success' type object. .. code:: python >>> def rescuable(arg: str) -> Result[str, str]: ... return Success(arg + 'b') ... >>> Failure('a').rescue(rescuable) == Success('ab') True """ return function(self._inner_value) def alt(self, function): """ Applies function to the error value. Applies 'function' to the contents of the '_Failure' instance and returns a new '_Failure' object containing the result. 'function' should accept a single "normal" (non-container) argument and return a non-container result. .. code:: python >>> def altable(arg: str) -> Result[str, str]: ... return arg + 'b' ... >>> Failure('a').alt(altable) == Failure('ab') True """ return _Failure(function(self._inner_value)) def value_or(self, default_value): """ Returns the value if we deal with '_Success' or default otherwise. .. code:: python >>> Failure(1).value_or(2) 2 """ return default_value def unwrap(self): """ Raises an exception, since it does not have a value inside. .. code:: python >>> Failure(1).unwrap() Traceback (most recent call last): ... returns.primitives.exceptions.UnwrapFailedError """ if isinstance(self._inner_value, Exception): raise UnwrapFailedError(self) from self._inner_value raise UnwrapFailedError(self) def failure(self): """ Unwraps inner error value from failed container. .. code:: python >>> Failure(1).failure() 1 """ return self._inner_value @final class _Success(Result[_ValueType, Any]): """ Represents a calculation which has succeeded and contains the result. Contains the computation value. Should not be used directly. """ _inner_value: _ValueType def __init__(self, inner_value: _ValueType) -> None: """Required for typing.""" BaseContainer.__init__(self, inner_value) # noqa: WPS609 def map(self, function): # noqa: A003 """ Applies function to the inner value. Applies 'function' to the contents of the '_Success' instance and returns a new '_Success' object containing the result. 'function' should accept a single "normal" (non-container) argument and return a non-container result. .. code:: python >>> def mappable(string: str) -> str: ... return string + 'b' ... >>> Success('a').map(mappable) == Success('ab') True """ return _Success(function(self._inner_value)) def bind(self, function): """ Applies 'function' to the result of a previous calculation. 'function' should accept a single "normal" (non-container) argument and return Result a '_Failure' or '_Success' type object. .. code:: python >>> def bindable(string: str) -> Result[str, str]: ... return Success(string + 'b') ... >>> Success('a').bind(bindable) == Success('ab') True """ return function(self._inner_value) def fix(self, function): """ Returns the '_Success' instance that was used to call the method. .. code:: python >>> def fixable(arg: str) -> str: ... return 'ab' ... >>> Success('a').fix(fixable) == Success('a') True """ return self def rescue(self, function): """ Returns the '_Success' instance that was used to call the method. .. code:: python >>> def rescuable(arg: str) -> Result[str, str]: ... return Success(arg + 'b') ... >>> Success('a').rescue(rescuable) == Success('a') True """ return self def alt(self, function): """ Returns the '_Success' instance that was used to call the method. .. code:: python >>> def altable(arg: str) -> Result[str, str]: ... return Success(arg + 'b') ... >>> Success('a').alt(altable) == Success('a') True """ return self def value_or(self, default_value): """ Returns the value if we deal with '_Success' or default otherwise. .. code:: python >>> Success(1).value_or(2) 1 """ return self._inner_value def unwrap(self): """ Returns the unwrapped value from the inside of this container. .. code:: python >>> Success(1).unwrap() 1 """ return self._inner_value def failure(self): """ Raises an exception, since it does not have an error inside. .. code:: python >>> Success(1).failure() Traceback (most recent call last): ... returns.primitives.exceptions.UnwrapFailedError """ raise UnwrapFailedError(self) def Success( # noqa: N802 inner_value: _ValueType, # type: ignore ) -> Result[_ValueType, NoReturn]: """Public unit function of protected `_Success` type.""" return _Success(inner_value) def Failure( # noqa: N802 inner_value: _ErrorType, # type: ignore ) -> Result[NoReturn, _ErrorType]: """Public unit function of protected `_Failure` type.""" return _Failure(inner_value) @overload def safe( # type: ignore function: Callable[..., Coroutine[_FirstType, _SecondType, _ValueType]], ) -> Callable[ ..., Coroutine[_FirstType, _SecondType, Result[_ValueType, Exception]], ]: """Case for async functions.""" @overload def safe( function: Callable[..., _ValueType], ) -> Callable[..., Result[_ValueType, Exception]]: """Case for regular functions.""" def safe(function): # noqa: C901 """ Decorator to covert exception throwing function to 'Result' container. Should be used with care, since it only catches 'Exception' subclasses. It does not catch 'BaseException' subclasses. Supports both async and regular functions. >>> @safe ... def might_raise(arg: int) -> float: ... return 1 / arg ... >>> might_raise(1) == Success(1.0) True >>> isinstance(might_raise(0), _Failure) True """ if iscoroutinefunction(function): async def decorator(*args, **kwargs): # noqa: WPS430 try: return Success(await function(*args, **kwargs)) except Exception as exc: return Failure(exc) else: def decorator(*args, **kwargs): # noqa: WPS430 try: return Success(function(*args, **kwargs)) except Exception as exc: return Failure(exc) return wraps(function)(decorator) PK!FHHreturns/unsafe.py# -*- coding: utf-8 -*- from typing import TypeVar from returns.io import IO _ValueType = TypeVar('_ValueType') def unsafe_perform_io(wrapped_in_io: IO[_ValueType]) -> _ValueType: """ Compatibility utility and escape mechanism from ``IO`` world. Just unwraps the internal value from :py:class:`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: - https://github.com/seddonym/import-linter .. code:: python >>> from returns.io import IO >>> unsafe_perform_io(IO(1)) 1 """ return wrapped_in_io._inner_value # noqa: WPS437 PK!q/   returns-0.11.0.dist-info/LICENSECopyright 2016-2019 dry-python organization Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. PK!HڽTUreturns-0.11.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!HzW'!returns-0.11.0.dist-info/METADATAZko8_L?4.,9m5si4-Fms"c,﹗,'}L0~XLE^Ry,?uS({ |kk􏎲伭*iSF^*5˶=ָI8S)JTJwٖc᷍*BօprX6r7n:S3dת0˴I~ֹ=;>Ju*n-ܴ@feeUWpSx+k4eTq6ITqmc+uXjt ?7$f\d V%p?fi a,З} wfܻ'^)d `r `R, s- PW+|X(T^J+N1<B^hpX(kl<~Ֆ6#ڄu[>tݜw?_vQBK Q&jH >KkgZTnr FJM@->P5:mc l䠱X(0K!Wq>xK-IBXZ##5 Q- ^҅ȝT$ IÓtY˴ҫ;=3@R;;,7!Зfk_|gGɀN/JڦlWNR_ uGik'0 I*7yFSuB4Y\ gx4Mf\I8ۉ'|1V/63$A~{d~PuTZNt?*ӕN'30т?'Ei]yS,C em6bm|H,|;D|"](ѧ{}ߌI Wc 5KHBA -{dՌGNN1M%G ]5<iWۖ1Qx` slUԢTN@2Sht"5-ay<{48knșSKeiKY.Kp/ o‘/ަ 2܌ņe-!| 쬨CeZ2h #ќa,%w\)]0,yzSP-7yl#(]hh)%ZvT& wgdxp9qw~Cp 97,7N@BLȊST8$@ط!gB'Kk)!t:نn a,$)<D_gS\GZbSR H?52^F1H}rC oyg)E!EV ¿b"ɏ6oaBI65 >@nWld)=h ŀZ""'Ü5q(L7}OKbǂ y%9?rAA*J;d& L6"MK3ˡ"4brϑÁzhHBa8q02.v''$ !򻣣nI .#˭5v,E^Y ]2ϓ]u"nn BEhh*}͵t}Hos D^i5:Ov7yk6cTfKGl 6K^G~׏t瘳8~hFjUg# j<Tv  AUcy]mC_ZT:<wHCD+ ΎvqJ0 W&%,DݝX}$yHxgAFl4{% d`pAT6EDR*r{[mgD=Wqh{1Yqpm(-f_& \ghچBI~Pd^or-G*I-B8K!o h;Zr]Gh WuKih,@E O`E^mt^ٌW\:5N> q>٫oB4]^C>٠v۰eR=VLnH9KLx! &"1s wowv*;*] 74.PC  ⢑ D#| Kw;hQ;Nj¤&Σ$CW