Spaces:
Running
Running
File size: 2,343 Bytes
6a0e448 |
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 |
import asyncio
import logging
import sys
from typing import Literal
from uuid import uuid4
from rich.logging import RichHandler
class StructuredLogger(logging.Logger):
async def stream_message(self, message: str) -> None:
"""Streams the message character by character asynchronously."""
try:
sys.stdout.write("\r") # Overwrite current line
for char in message:
sys.stdout.write(char)
sys.stdout.flush()
await asyncio.sleep(0.002)
sys.stdout.write("\n")
except Exception:
pass
def _log(
self,
level,
msg,
args,
exc_info=None,
extra=None,
stack_info=False,
stacklevel=1,
):
if extra is None:
extra = {}
json_fields = {
"logger_name": self.name,
"message": msg % args if args else msg,
}
exc_type, exc_value, exc_traceback = sys.exc_info()
if exc_type is not None:
json_fields["exception_class"] = exc_type.__name__
json_fields["exception_message"] = str(exc_value)
json_fields.update(extra)
super()._log(
level,
msg,
args,
exc_info,
{"json_fields": json_fields},
stack_info,
stacklevel + 1,
)
def create_logger(
name: str,
level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO",
detailed_name: bool = False,
) -> logging.Logger:
unique_name = f"{name}-{str(uuid4())[:8]}"
logger = logging.getLogger(unique_name)
logger.setLevel(level)
# Standard RichHandler for structured logs
rich_handler = RichHandler(
rich_tracebacks=True,
markup=True,
show_path=False,
show_time=False,
log_time_format="[%s]",
)
if detailed_name:
rich_handler.setFormatter(logging.Formatter("%(name)s:\n%(message)s"))
else:
rich_handler.setFormatter(logging.Formatter("-----\n%(message)s"))
logger.addHandler(rich_handler)
logger.propagate = False
return logger
# Set StructuredLogger as the default logger class
logging.setLoggerClass(StructuredLogger)
# Initialize logger
logger = create_logger(__name__, level="INFO")
|