85 lines
2.1 KiB
Python
85 lines
2.1 KiB
Python
import sys
|
|
from functools import wraps
|
|
from operator import attrgetter
|
|
from typing import Callable, Any
|
|
|
|
import loguru
|
|
from funcy import log_calls, log_enters, log_exits
|
|
|
|
logger = loguru.logger
|
|
logger.remove()
|
|
|
|
debug_logger = loguru.logger
|
|
debug_logger.add(
|
|
sink=sys.stderr,
|
|
format="<blue>{time:YYYY-MM-DD at HH:mm:ss}</blue> | <level>{level: <8}</level> | <cyan>{name}</cyan>: <level>{message}</level>",
|
|
level="DEBUG",
|
|
)
|
|
|
|
dev_logger = loguru.logger
|
|
dev_logger.add(
|
|
sink=sys.stderr,
|
|
format="<green>{time:YYYY-MM-DD at HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>: <level>{message}</level>",
|
|
level="DEBUG",
|
|
)
|
|
|
|
prod_logger = loguru.logger
|
|
prod_logger.add(
|
|
sink=sys.stderr,
|
|
format="<white>{time:YYYY-MM-DD at HH:mm:ss}</white> | <level>{level: <8}</level> | <cyan>{name}</cyan>: <level>{message}</level>",
|
|
level="INFO",
|
|
enqueue=True,
|
|
)
|
|
|
|
# logger.remove()
|
|
# logger.add(sink=sys.stderr, level="DEBUG", enqueue=True)
|
|
|
|
|
|
def __log(logger, level: str, enters=True, exits=True) -> Callable:
|
|
print_func = get_print_func(logger, level)
|
|
|
|
def dec():
|
|
if enters and exits:
|
|
fn = log_calls
|
|
elif enters:
|
|
fn = log_enters
|
|
elif exits:
|
|
fn = log_exits
|
|
else:
|
|
raise ValueError("Must log either enters or exits")
|
|
|
|
return fn(print_func=print_func)
|
|
|
|
def inner(func: Callable) -> Callable:
|
|
@dec()
|
|
@wraps(func)
|
|
def inner(*args, **kwargs) -> Any:
|
|
return func(*args, **kwargs)
|
|
|
|
return inner
|
|
|
|
return inner
|
|
|
|
|
|
def get_print_func(logger, level: str):
|
|
return attrgetter(level.lower())(logger)
|
|
|
|
|
|
def debug_log(level: str = "TRACE", enters=True, exits=True) -> Callable:
|
|
return __log(debug_logger, level, enters=enters, exits=exits)
|
|
|
|
|
|
def dev_log(level: str = "TRACE", enters=True, exits=True) -> Callable:
|
|
return __log(dev_logger, level, enters=enters, exits=exits)
|
|
|
|
|
|
def prod_log(level: str = "TRACE", enters=True, exits=True) -> Callable:
|
|
return __log(prod_logger, level, enters=enters, exits=exits)
|
|
|
|
|
|
def delay(fn, *args, **kwargs):
|
|
def inner():
|
|
return fn(*args, **kwargs)
|
|
|
|
return inner
|