loguru
loguru copied to clipboard
How can I retrieve loguru descriptive traceback?
I want to retrieve the loguru descriptive traceback when doing the serialized logging.
Is there is a function like traceback.format_exception(*error)
but from loguru how do I get the descriptive traceback.
Here is my current setup:
import json
import sys
import traceback
import loguru
from loguru import logger
PIPELINE_NAME = "test"
SERIALIZED_LOGGING = True
def serialize(record):
"""Serialize to our custom needs for structured logging.
With `logger.exception(msg)` all fields of `exception` key is populated.
With `logger.error(msg)` `traceback` field remains unpopulated.
Anything passed as kwargs will also be appended to the log record.
Additionally, if logger.x("@message", show_exception_value=False) will not
serialize exception's original value and will make it None. Useful in case of
BulkWriteError which throws whole duplicate document.
Args:
record: a dict record of loguru
Returns:
a JSON encoded str.
"""
error: loguru.RecordException = record["exception"]
error_by_default = sys.exc_info() # logger.error
pipeline: str | None = record["extra"].get("pipeline", None)
show_exception_value: bool = record["extra"].get("show_exception_value", True)
level: str = record["level"].name
extra = record["extra"]
extra.update({"pipeline": pipeline})
if error: # only set when exception.
exception = {
"type": error.type.__name__,
"value": str(error.value) if show_exception_value else None,
"traceback": "".join(traceback.format_exception(*error)),
}
elif error_by_default[0]: # whenever error occurs
_type, _value, _ = sys.exc_info()
exception = {
"type": _type.__name__,
"value": str(_value) if show_exception_value else None,
"traceback": None,
}
else:
exception = None
to_serialize = {
"level": level,
# "time": record["time"].strftime("%d-%m-%YT%H:%M:%SZ"),
"message": record["message"],
"pipeline": pipeline,
"exception": exception,
}
to_serialize.update(extra)
return json.dumps(to_serialize)
def patching(record):
"""Patch the logger."""
record["extra"]["serialized"] = serialize(record)
def get_contextualized_logger(
pipeline_name: str = PIPELINE_NAME, default_logger=logger
):
"""Generates a contextualized logger with pipeline_name (P1/P2/P3)."""
if not SERIALIZED_LOGGING:
return default_logger
default_logger.remove()
default_logger = default_logger.patch(patching)
default_logger.add(
sink=sys.stderr,
serialize=False, # custom serialization requires this to be False
backtrace=True,
diagnose=False, # enable for debugging; shows runtime vals of call func
level="INFO",
format="{extra[serialized]}",
)
return default_logger