Spaces:
Sleeping
Sleeping
File size: 3,023 Bytes
5564ecb 8c09fcd 5564ecb |
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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import inspect
import logging
import os
import sys
from asgi_correlation_id import correlation_id
from loguru import logger
from core import path_conf
from core.conf import settings
class InterceptHandler(logging.Handler):
def emit(self, record: logging.LogRecord):
try:
level = logger.level(record.levelname).name
except ValueError:
level = record.levelno
frame, depth = inspect.currentframe(), 0
while frame and (depth == 0 or frame.f_code.co_filename == logging.__file__):
frame = frame.f_back
depth += 1
logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
def setup_logging() -> None:
"""
- https://github.com/benoitc/gunicorn/issues/1572#issuecomment-638391953
- https://github.com/pawamoy/pawamoy.github.io/issues/17
"""
logging.root.handlers = [InterceptHandler()]
logging.root.setLevel(settings.LOG_STD_LEVEL)
for name in logging.root.manager.loggerDict.keys():
logging.getLogger(name).handlers = []
if 'uvicorn.access' in name or 'watchfiles.main' in name:
logging.getLogger(name).propagate = False
else:
logging.getLogger(name).propagate = True
# Debug log handlers
# logging.debug(f'{logging.getLogger(name)}, {logging.getLogger(name).propagate}')
# https://github.com/snok/asgi-correlation-id/issues/7
def correlation_id_filter(record):
cid = correlation_id.get(settings.LOG_CID_DEFAULT_VALUE)
record['correlation_id'] = cid[: settings.LOG_CID_UUID_LENGTH]
return record
logger.remove()
logger.configure(
handlers=[
{
'sink': sys.stdout,
'level': settings.LOG_STD_LEVEL,
'filter': lambda record: correlation_id_filter(record),
'format': settings.LOG_STD_FORMAT,
}
]
)
def set_custom_logfile() -> None:
log_path = path_conf.LOG_DIR
if not os.path.exists(log_path):
os.mkdir(log_path)
log_access_file = os.path.join(log_path, settings.LOG_ACCESS_FILENAME)
log_error_file = os.path.join(log_path, settings.LOG_ERROR_FILENAME)
# https://loguru.readthedocs.io/en/stable/api/logger.html#loguru._logger.Logger.add
log_config = {
'format': settings.LOG_FILE_FORMAT,
'enqueue': True,
'rotation': '5 MB',
'retention': '7 days',
'compression': 'tar.gz',
}
logger.add(
str(log_access_file),
level=settings.LOG_ACCESS_FILE_LEVEL,
filter=lambda record: record['level'].no <= 25,
backtrace=False,
diagnose=False,
**log_config,
)
logger.add(
str(log_error_file),
level=settings.LOG_ERROR_FILE_LEVEL,
filter=lambda record: record['level'].no >= 30,
backtrace=True,
diagnose=True,
**log_config,
)
log = logger
|