2018-10-11 05:26:00 +00:00
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Callable, Generic, TypeVar
|
|
|
|
|
2018-10-12 18:25:19 +00:00
|
|
|
from .monad import Monad
|
|
|
|
|
2018-10-11 05:26:00 +00:00
|
|
|
T = TypeVar("T")
|
|
|
|
S = TypeVar("S")
|
|
|
|
E = TypeVar("E")
|
|
|
|
|
|
|
|
|
|
|
|
class Result(Monad[T], Generic[T, E]):
|
2018-10-12 18:24:03 +00:00
|
|
|
def __init__(self) -> None: # pragma: no cover
|
2018-10-11 05:26:00 +00:00
|
|
|
raise NotImplementedError
|
|
|
|
|
|
|
|
@classmethod
|
2018-10-12 01:06:23 +00:00
|
|
|
def pure(cls, value: T) -> Result[T, E]:
|
2018-10-11 05:26:00 +00:00
|
|
|
return Ok(value)
|
|
|
|
|
|
|
|
def bind(self, function: Callable[[T], Result[S, E]]) -> Result[S, E]:
|
|
|
|
if isinstance(self, Ok):
|
|
|
|
return function(self.value)
|
|
|
|
elif isinstance(self, Err):
|
|
|
|
new: Result[S, E] = Err(self.err)
|
|
|
|
return new
|
2018-10-12 18:55:19 +00:00
|
|
|
else: # pragma: no cover
|
2018-10-11 05:26:00 +00:00
|
|
|
raise TypeError
|
|
|
|
|
2018-10-12 01:06:23 +00:00
|
|
|
def map(self, function: Callable[[T], S]) -> Result[S, E]:
|
2018-10-11 05:26:00 +00:00
|
|
|
if isinstance(self, Ok):
|
2018-10-12 01:06:23 +00:00
|
|
|
return Result.pure(function(self.value))
|
2018-10-11 05:26:00 +00:00
|
|
|
elif isinstance(self, Err):
|
|
|
|
new: Result[S, E] = Err(self.err)
|
|
|
|
return new
|
2018-10-12 18:55:19 +00:00
|
|
|
else: # pragma: no cover
|
2018-10-11 05:26:00 +00:00
|
|
|
raise TypeError
|
|
|
|
|
2018-10-12 18:09:16 +00:00
|
|
|
def apply(self, functor: Result[Callable[[T], S], E]) -> Result[S, E]:
|
|
|
|
if isinstance(functor, Ok):
|
|
|
|
return self.map(functor.value)
|
|
|
|
elif isinstance(functor, Err):
|
|
|
|
new: Result[S, E] = Err(functor.err)
|
|
|
|
return new
|
2018-10-12 18:55:19 +00:00
|
|
|
else: # pragma: no cover
|
2018-10-12 18:09:16 +00:00
|
|
|
raise TypeError
|
|
|
|
|
2018-10-11 05:26:00 +00:00
|
|
|
def withDefault(self, default: T) -> T:
|
|
|
|
if isinstance(self, Ok):
|
|
|
|
return self.value
|
|
|
|
else:
|
|
|
|
return default
|
|
|
|
|
2018-12-03 21:16:04 +00:00
|
|
|
@classmethod
|
2018-12-06 18:15:26 +00:00
|
|
|
def fromMaybe(cls, m: Maybe[T], error: E) -> Result[T, E]:
|
2018-12-03 21:16:04 +00:00
|
|
|
return m.map(Result.pure).withDefault(Err(error))
|
|
|
|
|
|
|
|
def toMaybe(self) -> Maybe[T]:
|
|
|
|
return self.map(Maybe.pure).withDefault(Nothing())
|
|
|
|
|
2018-10-11 05:26:00 +00:00
|
|
|
__rshift__ = bind
|
2018-10-12 01:06:23 +00:00
|
|
|
__mul__ = __rmul__ = map
|
2018-10-11 05:26:00 +00:00
|
|
|
|
|
|
|
|
|
|
|
class Ok(Result[T, E]):
|
|
|
|
def __init__(self, value: T) -> None:
|
|
|
|
self.value = value
|
|
|
|
|
2018-10-11 16:15:52 +00:00
|
|
|
def __eq__(self, other: object):
|
|
|
|
return isinstance(other, Ok) and self.value == other.value
|
|
|
|
|
2018-10-12 18:24:03 +00:00
|
|
|
def __repr__(self) -> str: # pragma: no cover
|
2018-10-11 05:26:00 +00:00
|
|
|
return f"<Ok {self.value}>"
|
|
|
|
|
|
|
|
|
|
|
|
class Err(Result[T, E]):
|
|
|
|
def __init__(self, err: E) -> None:
|
|
|
|
self.err = err
|
|
|
|
|
2018-10-11 16:15:52 +00:00
|
|
|
def __eq__(self, other: object):
|
|
|
|
return isinstance(other, Err) and self.err == other.err
|
|
|
|
|
2018-10-12 18:24:03 +00:00
|
|
|
def __repr__(self) -> str: # pragma: no cover
|
2018-10-11 05:26:00 +00:00
|
|
|
return f"<Err {self.err}>"
|
|
|
|
|
|
|
|
|
|
|
|
def safe(function: Callable[..., T]) -> Callable[..., Result[T, Exception]]:
|
|
|
|
"""Wraps a function that may raise an exception.
|
|
|
|
|
|
|
|
e.g.:
|
|
|
|
@safe
|
|
|
|
def bad() -> int:
|
|
|
|
raise Exception("oops")
|
|
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
def wrapped(*args, **kwargs) -> Result[T, Exception]:
|
|
|
|
try:
|
|
|
|
return Ok(function(*args, **kwargs))
|
|
|
|
except Exception as e:
|
|
|
|
return Err(e)
|
|
|
|
|
|
|
|
return wrapped
|
2018-12-06 18:14:40 +00:00
|
|
|
|
|
|
|
|
|
|
|
# Import Maybe last to avoid a circular import error
|
|
|
|
from .maybe import Maybe, Nothing
|