150 lines
4.9 KiB
Python
150 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
# IMPORTS
|
|
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
|
|
from __future__ import annotations;
|
|
|
|
from src.thirdparty.code import *;
|
|
from src.thirdparty.misc import *;
|
|
from src.thirdparty.run import *;
|
|
from src.thirdparty.types import *;
|
|
|
|
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
# EXPORTS
|
|
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
|
|
__all__ = [
|
|
'CallResult',
|
|
'CallError',
|
|
'run_safely',
|
|
];
|
|
|
|
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
# CONSTANTS
|
|
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
|
|
# local usage only
|
|
T = TypeVar('T');
|
|
V = TypeVar('V');
|
|
E = TypeVar('E', bound=list);
|
|
ARGS = ParamSpec('ARGS');
|
|
|
|
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
# CLASS Trace for debugging only!
|
|
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
|
|
@dataclass
|
|
class CallResult(): # pragma: no cover
|
|
'''
|
|
An auxiliary class which keeps track of the latest return value during calls.
|
|
'''
|
|
action_taken: bool = field(default=False);
|
|
message: Optional[Any] = field(default=None);
|
|
|
|
@dataclass
|
|
class CallErrorRaw(): # pragma: no cover
|
|
timestamp: str = field();
|
|
tag: str = field();
|
|
errors: List[str] = field(default_factory=list);
|
|
|
|
class CallError(CallErrorRaw):
|
|
'''
|
|
An auxiliary class which keeps track of potentially multiple errors during calls.
|
|
'''
|
|
timestamp: str;
|
|
tag: str;
|
|
errors: List[str];
|
|
|
|
def __init__(self, tag: str, err: Any = Nothing()):
|
|
self.timestamp = str(datetime.now());
|
|
self.tag = tag;
|
|
self.errors = [];
|
|
if isinstance(err, list):
|
|
for e in err:
|
|
self.append(e);
|
|
else:
|
|
self.append(err);
|
|
|
|
def __len__(self) -> int:
|
|
return len(self.errors);
|
|
|
|
def append(self, e: Any):
|
|
if isinstance(e, Nothing):
|
|
return;
|
|
if isinstance(e, Some):
|
|
e = e.unwrap();
|
|
self.errors.append(str(e));
|
|
|
|
def extend(self, E: CallError):
|
|
self.errors.extend(E.errors);
|
|
|
|
def __repr__(self) -> str:
|
|
return f'CallError(tag=\'{self.tag}\', errors={self.errors})';
|
|
|
|
def __str__(self) -> str:
|
|
return self.__repr__();
|
|
|
|
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
# DECORATOR - forces methods to run safely
|
|
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
|
|
def run_safely(tag: Union[str, None] = None, error_message: Union[str, None] = None):
|
|
'''
|
|
Creates a decorator for an action to perform it safely.
|
|
|
|
@inputs (parameters)
|
|
- `tag` - optional string to aid error tracking.
|
|
- `error_message` - optional string for an error message.
|
|
|
|
### Example usage ###
|
|
```py
|
|
@run_safely(tag='recognise int', error_message='unrecognise string')
|
|
def action1(x: str) -> Result[int, CallError]:
|
|
return Ok(int(x));
|
|
|
|
assert action1('5') == Ok(5);
|
|
result = action1('not a number');
|
|
assert isinstance(result, Err);
|
|
err = result.unwrap_err();
|
|
assert isinstance(err, CallError);
|
|
assert err.tag == 'recognise int';
|
|
assert err.errors == ['unrecognise string'];
|
|
|
|
@run_safely('recognise int')
|
|
def action2(x: str) -> Result[int, CallError]:
|
|
return Ok(int(x));
|
|
|
|
assert action2('5') == Ok(5);
|
|
result = action2('not a number');
|
|
assert isinstance(result, Err);
|
|
err = result.unwrap_err();
|
|
assert isinstance(err, CallError);
|
|
assert err.tag == 'recognise int';
|
|
assert len(err.errors) == 1;
|
|
```
|
|
NOTE: in the second example, err.errors is a list containing
|
|
the stringified Exception generated when calling `int('not a number')`.
|
|
'''
|
|
def dec(action: Callable[ARGS, Result[V, CallError]]) -> Callable[ARGS, Result[V, CallError]]:
|
|
'''
|
|
Wraps action with return type Result[..., CallError],
|
|
so that it is performed safely a promise,
|
|
catching any internal exceptions as an Err(...)-component of the Result.
|
|
'''
|
|
@wraps(action)
|
|
def wrapped_action(*_, **__) -> Result[V, CallError]:
|
|
# NOTE: intercept Exceptions first, then flatten:
|
|
return Result.of(lambda: action(*_, **__)) \
|
|
.or_else(
|
|
lambda err: Err(CallError(
|
|
tag = tag or action.__name__,
|
|
err = error_message or err
|
|
))
|
|
) \
|
|
.and_then(lambda V: V);
|
|
return wrapped_action;
|
|
return dec;
|