Diff of /catenets/logger.py [000000] .. [6ac965]

Switch to side-by-side view

--- a
+++ b/catenets/logger.py
@@ -0,0 +1,103 @@
+# stdlib
+import logging
+import os
+from typing import Any, Callable, NoReturn, TextIO, Union
+
+# third party
+from loguru import logger
+
+LOG_FORMAT = "[{time}][{process.id}][{level}] {message}"
+
+logger.remove()
+DEFAULT_SINK = "catenets_{time}.log"
+
+
+def remove() -> None:
+    logger.remove()
+
+
+def add(
+    sink: Union[None, str, os.PathLike, TextIO, logging.Handler] = None,
+    level: str = "ERROR",
+) -> None:
+    sink = DEFAULT_SINK if sink is None else sink
+    try:
+        logger.add(
+            sink=sink,
+            format=LOG_FORMAT,
+            enqueue=True,
+            colorize=False,
+            diagnose=True,
+            backtrace=True,
+            rotation="10 MB",
+            retention="1 day",
+            level=level,
+        )
+    except BaseException:
+        logger.add(
+            sink=sink,
+            format=LOG_FORMAT,
+            colorize=False,
+            diagnose=True,
+            backtrace=True,
+            level=level,
+        )
+
+
+def traceback_and_raise(e: Any, verbose: bool = False) -> NoReturn:
+    try:
+        if verbose:
+            logger.opt(lazy=True).exception(e)
+        else:
+            logger.opt(lazy=True).critical(e)
+    except BaseException as ex:
+        logger.debug("failed to print exception", ex)
+    if not issubclass(type(e), Exception):
+        e = Exception(e)
+    raise e
+
+
+def create_log_and_print_function(level: str) -> Callable:
+    def log_and_print(*args: Any, **kwargs: Any) -> None:
+        try:
+            method = getattr(logger.opt(lazy=True), level, None)
+            if method is not None:
+                method(*args, **kwargs)
+            else:
+                logger.debug(*args, **kwargs)
+        except BaseException as e:
+            msg = f"failed to log exception. {e}"
+            try:
+                logger.debug(msg)
+            except Exception as e:
+                print(f"{msg}. {e}")
+
+    return log_and_print
+
+
+def traceback(*args: Any, **kwargs: Any) -> None:
+    return create_log_and_print_function(level="exception")(*args, **kwargs)
+
+
+def critical(*args: Any, **kwargs: Any) -> None:
+    return create_log_and_print_function(level="critical")(*args, **kwargs)
+
+
+def error(*args: Any, **kwargs: Any) -> None:
+    return create_log_and_print_function(level="error")(*args, **kwargs)
+
+
+def warning(*args: Any, **kwargs: Any) -> None:
+    return create_log_and_print_function(level="warning")(*args, **kwargs)
+
+
+def info(*args: Any, **kwargs: Any) -> None:
+    return create_log_and_print_function(level="info")(*args, **kwargs)
+
+
+def debug(*args: Any, **kwargs: Any) -> None:
+    return create_log_and_print_function(level="debug")(*args, **kwargs)
+
+
+def trace(*args: Any, **kwargs: Any) -> None:
+    return create_log_and_print_function(level="trace")(*args, **kwargs)