Spaces:
Running
Running
File size: 1,433 Bytes
9e6aa30 |
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 |
import contextvars
import logging
from functools import wraps
from typing import Callable
logging_uuid = contextvars.ContextVar("uuid")
formatter = '%(asctime)s | %(uuid)s [%(pathname)s:%(module)s %(lineno)d] %(levelname)s | %(message)s'
formatter_without_uuid = '%(asctime)s | [%(pathname)s:%(module)s %(lineno)d] %(levelname)s | %(message)s'
loggingType = logging.CRITICAL | logging.ERROR | logging.WARNING | logging.INFO | logging.DEBUG
def change_logging(level_log: loggingType = logging.INFO, formatter_log: str = formatter) -> None:
try:
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)
except Exception as e:
print(f"change_logging exception:{e}, reset logging...")
formatter_log = formatter_without_uuid
logging.basicConfig(level=level_log, format=formatter_log, force=True)
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
|