2024-08-30 16:46:37 +00:00
|
|
|
import logging
|
|
|
|
from logging.handlers import MemoryHandler
|
|
|
|
from collections import deque
|
|
|
|
|
|
|
|
logs = None
|
|
|
|
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
|
|
|
|
|
|
|
|
|
|
|
def get_logs():
|
|
|
|
return "\n".join([formatter.format(x) for x in logs])
|
|
|
|
|
|
|
|
|
2024-10-04 14:05:34 +00:00
|
|
|
def setup_logger(log_level: str = 'INFO', capacity: int = 300):
|
2024-08-30 16:46:37 +00:00
|
|
|
global logs
|
|
|
|
if logs:
|
|
|
|
return
|
|
|
|
|
|
|
|
# Setup default global logger
|
|
|
|
logger = logging.getLogger()
|
2024-10-04 14:05:34 +00:00
|
|
|
logger.setLevel(log_level)
|
2024-08-30 16:46:37 +00:00
|
|
|
|
|
|
|
stream_handler = logging.StreamHandler()
|
|
|
|
stream_handler.setFormatter(logging.Formatter("%(message)s"))
|
|
|
|
logger.addHandler(stream_handler)
|
|
|
|
|
|
|
|
# Create a memory handler with a deque as its buffer
|
|
|
|
logs = deque(maxlen=capacity)
|
|
|
|
memory_handler = MemoryHandler(capacity, flushLevel=logging.INFO)
|
|
|
|
memory_handler.buffer = logs
|
|
|
|
memory_handler.setFormatter(formatter)
|
|
|
|
logger.addHandler(memory_handler)
|