master > master: code py - refactored code mit Endpunkten

code/python/src/core/calls.py

@@ -0,0 +1,149 @@
+#!/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;

code/python/src/core/log.py

@@ -5,60 +5,93 @@
 # IMPORTS
 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-#
+from src.thirdparty.code import *;
+from src.thirdparty.log import *;
+from src.thirdparty.types import *;
+
+from src.core.calls import *;
 
 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 # EXPORTS
 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 __all__ = [
+    'LOG_LEVELS',
+    'configure_logging',
     'log_info',
     'log_debug',
     'log_warn',
     'log_error',
     'log_fatal',
+    'log_result',
+    'log_dev',
 ];
 
 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-# METHODS logging
+# CONSTANTS
 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-def log_info(*text: str):
-    '''
-    Prints an info message
-    '''
-    for line in text:
-        print("[\x1b[94;1mINFO\x1b[0m] {}".format(line));
+_LOGGING_DEBUG_FILE: str = 'logs/debug.log';
+
+class LOG_LEVELS(Enum): # pragma: no cover
+    INFO  = logging.INFO;
+    DEBUG = logging.DEBUG;
+
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+# METHODS
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+def configure_logging(level: LOG_LEVELS): # pragma: no cover
+    logging.basicConfig(
+        format = '[\x1b[1m%(levelname)s\x1b[0m] %(message)s',
+        level  = level.value,
+    );
     return;
 
-def log_debug(*text: str):
-    '''
-    Prints a debug message
-    '''
-    for line in text:
-        print("[\x1b[96;1mDEBUG\x1b[95;0m] {}".format(line));
-    return;
+def log_debug(*messages: Any):
+    logging.debug(*messages);
 
-def log_warn(*text: str):
-    '''
-    Prints a warning message
-    '''
-    for line in text:
-        print("[\x1b[93;1mWARNING\x1b[0m] {}".format(line));
-    return;
+def log_info(*messages: Any):
+    logging.info(*messages);
 
-def log_error(*text: str):
-    '''
-    Prints an error message
-    '''
-    for line in text:
-        print("[\x1b[91;1mERROR\x1b[0m] {}".format(line));
-    return;
+def log_warn(*messages: Any):
+    logging.warning(*messages);
 
-def log_fatal(*text: str):
-    '''
-    Prints a fatal error message + crashes
-    '''
-    for line in text:
-        print("[\x1b[91;1mFATAL\x1b[0m] {}".format(line));
+def log_error(*messages: Any):
+    logging.error(*messages);
+
+def log_fatal(*messages: Any):
+    logging.fatal(*messages);
     exit(1);
+
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+# Special Methods
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+def log_result(result: Result[CallResult, CallError], debug: bool = False):
+    '''
+    Logs safely encapsulated result of call as either debug/info or error.
+
+    @inputs
+    - `result` - the result of the call.
+    - `debug = False` (default) - if the result is okay, will be logged as an INFO message.
+    - `debug = True` - if the result is okay, will be logged as a DEBUG message.
+    '''
+    if isinstance(result, Ok):
+        value = result.unwrap();
+        if debug:
+            log_debug(asdict(value));
+        else:
+            log_info(asdict(value));
+    else:
+        err = result.unwrap_err();
+        log_error(asdict(err));
+    return;
+
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+# DEBUG LOGGING FOR DEVELOPMENT
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+def log_dev(*messages: Any): # pragma: no cover
+    with open(_LOGGING_DEBUG_FILE, 'a') as fp:
+        print(*messages, file=fp);