1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111
|
"""
An interface for types that do ``IO`` and can fail.
It is a base interface for both sync and async ``IO`` stacks.
"""
from __future__ import annotations
from abc import abstractmethod
from collections.abc import Callable
from typing import TYPE_CHECKING, TypeVar
from typing_extensions import Never
from returns.interfaces.specific import io, result
from returns.primitives.hkt import KindN
if TYPE_CHECKING:
from returns.io import IO, IOResult # noqa: WPS433
from returns.result import Result # noqa: WPS433
_FirstType = TypeVar('_FirstType')
_SecondType = TypeVar('_SecondType')
_ThirdType = TypeVar('_ThirdType')
_UpdatedType = TypeVar('_UpdatedType')
_ValueType = TypeVar('_ValueType')
_ErrorType = TypeVar('_ErrorType')
_IOResultLikeType = TypeVar('_IOResultLikeType', bound='IOResultLikeN')
class IOResultLikeN(
io.IOLikeN[_FirstType, _SecondType, _ThirdType],
result.ResultLikeN[_FirstType, _SecondType, _ThirdType],
):
"""
Base type for types that look like ``IOResult`` but cannot be unwrapped.
Like ``FutureResult`` or ``RequiresContextIOResult``.
"""
__slots__ = ()
@abstractmethod
def bind_ioresult(
self: _IOResultLikeType,
function: Callable[[_FirstType], IOResult[_UpdatedType, _SecondType]],
) -> KindN[_IOResultLikeType, _UpdatedType, _SecondType, _ThirdType]:
"""Runs ``IOResult`` returning function over a container."""
@abstractmethod
def compose_result(
self: _IOResultLikeType,
function: Callable[
[Result[_FirstType, _SecondType]],
KindN[_IOResultLikeType, _UpdatedType, _SecondType, _ThirdType],
],
) -> KindN[_IOResultLikeType, _UpdatedType, _SecondType, _ThirdType]:
"""Allows to compose the underlying ``Result`` with a function."""
@classmethod
@abstractmethod
def from_ioresult(
cls: type[_IOResultLikeType],
inner_value: IOResult[_ValueType, _ErrorType],
) -> KindN[_IOResultLikeType, _ValueType, _ErrorType, _ThirdType]:
"""Unit method to create new containers from ``IOResult`` type."""
@classmethod
@abstractmethod
def from_failed_io(
cls: type[_IOResultLikeType],
inner_value: IO[_ErrorType],
) -> KindN[_IOResultLikeType, _FirstType, _ErrorType, _ThirdType]:
"""Unit method to create new containers from failed ``IO``."""
#: Type alias for kinds with two type arguments.
IOResultLike2 = IOResultLikeN[_FirstType, _SecondType, Never]
#: Type alias for kinds with three type arguments.
IOResultLike3 = IOResultLikeN[_FirstType, _SecondType, _ThirdType]
class IOResultBasedN(
IOResultLikeN[_FirstType, _SecondType, _ThirdType],
io.IOBasedN[_FirstType, _SecondType, _ThirdType],
result.UnwrappableResult[
_FirstType,
_SecondType,
_ThirdType,
# Unwraps:
'IO[_FirstType]',
'IO[_SecondType]',
],
):
"""
Base type for real ``IOResult`` types.
Can be unwrapped.
"""
__slots__ = ()
#: Type alias for kinds with two type arguments.
IOResultBased2 = IOResultBasedN[_FirstType, _SecondType, Never]
#: Type alias for kinds with three type arguments.
IOResultBased3 = IOResultBasedN[_FirstType, _SecondType, _ThirdType]
|