File size: 1,801 Bytes
0914710 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 |
import contextvars
import logging
from functools import wraps
from typing import Callable, Tuple
logging_uuid = contextvars.ContextVar("uuid")
default_formatter = '%(asctime)s | %(uuid)s [%(pathname)s:%(module)s %(lineno)d] %(levelname)s | %(message)s'
loggingType = logging.CRITICAL | logging.ERROR | logging.WARNING | logging.INFO | logging.DEBUG
def setup_logging(
debug: bool = False, formatter: str = default_formatter, name: str = "logger"
) -> Tuple[logging, contextvars.ContextVar]:
"""
Create a logging instance with log string formatter.
Args:
debug: logging debug argument
formatter: log string formatter
name: logger name
Returns:
Logger
"""
old_factory = logging.getLogRecordFactory()
def record_factory(*args, **kwargs):
record = old_factory(*args, **kwargs)
record.uuid = logging_uuid.get("uuid")
if isinstance(record.msg, str):
record.msg = record.msg.replace("\\", "\\\\").replace("\n", "\\n")
return record
logging.setLogRecordFactory(record_factory)
logging.basicConfig(level=logging.DEBUG, format=default_formatter, force=True)
logger = logging.getLogger(name=name)
# create a console handler
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
# create formatter and add to the console
formatter = logging.Formatter(formatter)
ch.setFormatter(formatter)
# add the console handler to logger
logger.addHandler(ch)
return logger, logging_uuid
def set_uuid_logging(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
import uuid
current_uuid = f"{uuid.uuid4()}"
logging_uuid.set(current_uuid)
return func(*args, **kwargs)
return wrapper
|