File size: 4,708 Bytes
dbb04e4 | 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 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | """
MnemoCore Logging Configuration
================================
Centralized logging configuration using loguru.
Provides:
- configure_logging(): Setup function called at application startup
- JSON log format when LOG_FORMAT=json environment variable is set
- Consistent logging across all modules
Usage:
from mnemocore.core.logging_config import configure_logging, get_logger
# At application startup:
configure_logging(level="INFO", json_format=False)
# In modules:
logger = get_logger(__name__)
logger.info("Message")
"""
from __future__ import annotations
import os
import sys
from typing import Optional
from loguru import logger
# Remove default handler
logger.remove()
# Track if logging has been configured
_CONFIGURED = False
def configure_logging(
level: str = "INFO",
json_format: Optional[bool] = None,
*,
sink: Optional[str] = None,
) -> None:
"""
Configure loguru logging for MnemoCore.
Args:
level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL).
json_format: If True, use JSON format. If None, check LOG_FORMAT env var.
sink: Optional file path for log output. If None, logs to stderr.
Environment:
LOG_FORMAT: Set to "json" to enable JSON formatted logs.
LOG_LEVEL: Override log level if not specified.
"""
global _CONFIGURED
# Check environment for JSON format
if json_format is None:
json_format = os.environ.get("LOG_FORMAT", "").lower() == "json"
# Check environment for log level
if level is None:
level = os.environ.get("LOG_LEVEL", "INFO")
# Remove existing handlers
logger.remove()
# Determine sink
log_sink = sink if sink else sys.stderr
if json_format:
# JSON format for production/cloud logging
format_str = (
'{{"timestamp": "{{time:YYYY-MM-DDTHH:mm:ss.SSSZ}}", '
'"level": "{{level}}", '
'"logger": "{{name}}", '
'"function": "{{function}}", '
'"line": {{line}}, '
'"message": "{{message}}", '
'"exception": "{{exception}}"}}'
)
else:
# Human-readable format for development
format_str = (
"<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
"<level>{level: <8}</level> | "
"<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | "
"<level>{message}</level>"
)
# Add handler
logger.add(
log_sink,
level=level.upper(),
format=format_str,
colorize=not json_format and sink is None,
enqueue=True, # Thread-safe
backtrace=True,
diagnose=True,
)
# Intercept standard logging
_intercept_standard_logging(level)
_CONFIGURED = True
logger.debug(f"Logging configured: level={level}, json_format={json_format}")
def _intercept_standard_logging(level: str) -> None:
"""
Intercept standard library logging and redirect to loguru.
This ensures that logs from third-party libraries using stdlib logging
are also handled by loguru.
"""
import logging
class InterceptHandler(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
try:
log_level = logger.level(record.levelname).name
except ValueError:
log_level = record.levelno
frame, depth = logging.currentframe(), 2
while frame.f_code.co_filename == logging.__file__:
frame = frame.f_back # type: ignore
depth += 1
logger.opt(depth=depth, exception=record.exc_info).log(
log_level, record.getMessage()
)
# Configure root logger to use our handler
logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)
# Set levels for common noisy loggers
for logger_name in ["uvicorn", "uvicorn.error", "uvicorn.access"]:
logging.getLogger(logger_name).setLevel(level.upper())
def get_logger(name: str = __name__):
"""
Get a logger instance bound to the specified module name.
Args:
name: Module name (typically __name__).
Returns:
A loguru logger instance bound to the module.
"""
# Ensure logging is configured with defaults if not already done
if not _CONFIGURED:
configure_logging()
return logger.bind(name=name)
# Module-level logger for convenience
__all__ = ["configure_logging", "get_logger", "logger"]
|