import json
import sys
from loguru import logger
from loguru._handler import Handler
from loguru._recattrs import RecordException
from config import CONFIG, ROOT_PATH
from config.argparse_info import args_dict
ENV = args_dict["env"]
APP = CONFIG.get("project_config", "app")
class Logging:
def __init__(self, filename=APP):
hashtag if ENV != "prod":
hashtag return
logger.remove()
log_path = f"{ROOT_PATH}/log_out/{filename}.log"
hashtag if not os.path.exists(log_path):
hashtag with open(log_path, "w", encoding="utf8") as f:
hashtag ...
Handler._serialize_record = staticmethod(self._serialize_record)
logger.add(sys.stderr,
serialize=True
hashtag format="{time} - {level} - {message}"
)
logger.add(log_path,
rotation=CONFIG.get('Log', 'rotation'),
encoding="utf-8",
hashtag enqueue=True,
retention=CONFIG.get('Log', 'retention'),
compression=CONFIG.get('Log', 'compression'),
format=CONFIG.get("Log", "format"),
level="INFO",
hashtag diagnose=True,
serialize=True
)
def _serialize_record(text: str, record: dict):
exception: RecordException = record["exception"]
serializable = {
hashtag "text": text,
hashtag "record": {
hashtag "elapsed": {
hashtag "repr": record["elapsed"],
hashtag "seconds": record["elapsed"].total_seconds(),
hashtag },
hashtag "exception": exception,
hashtag "extra": record["extra"],
hashtag "file": {"name": record["file"].name, "path": record["file"].path},
"function": record["function"],
hashtag "level": {
hashtag "icon": record["level"].icon,
hashtag "name": record["level"].name,
hashtag "no": record["level"].no,
hashtag },
hashtag "line": record["line"],
"message": record["message"],
hashtag "module": record["module"],
hashtag "name": record["name"],
"process": {
"id": record["process"].id,
"name": record["process"].name
},
"thread": {
"id": record["thread"].id,
"name": record["thread"].name
},
"time": record["time"].strftime('%Y-%m-%d %H:%M:%S'),
hashtag "time": {
hashtag "repr": record["time"],
hashtag "timestamp": record["time"].timestamp()
hashtag },
hashtag },
}
return json.dumps(serializable, default=str, ensure_ascii=False) + "\n"
def info(msg):
return logger.info(msg)
def debug(msg):
return logger.debug(msg)
def warning(msg):
return logger.warning(msg)
def error(msg):
return logger.error(msg)
Is it related to the number of processes I have?
