Spaces:
Sleeping
Sleeping
File size: 2,275 Bytes
409c17a |
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 |
"""
Core - Structured Logging
JSON structured logging with correlation IDs.
"""
import logging
import sys
from contextvars import ContextVar
from datetime import datetime
from typing import Any, Dict
import structlog
# Correlation ID context variable
correlation_id_var: ContextVar[str] = ContextVar("correlation_id", default="")
def get_correlation_id() -> str:
"""Get current correlation ID"""
return correlation_id_var.get()
def set_correlation_id(correlation_id: str) -> None:
"""Set correlation ID for current context"""
correlation_id_var.set(correlation_id)
def add_correlation_id(logger: Any, method_name: str, event_dict: Dict) -> Dict:
"""Add correlation ID to log event"""
event_dict["correlation_id"] = get_correlation_id()
return event_dict
def add_service_info(logger: Any, method_name: str, event_dict: Dict) -> Dict:
"""Add service information to log event"""
event_dict["service"] = "rag-onboarding-backend"
event_dict["version"] = "1.0.0"
return event_dict
def setup_logging(log_level: str = "INFO") -> None:
"""Setup structured logging"""
# Configure structlog
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.stdlib.filter_by_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.processors.StackInfoRenderer(),
add_correlation_id,
add_service_info,
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
structlog.processors.JSONRenderer(),
],
wrapper_class=structlog.stdlib.BoundLogger,
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
# Configure standard logging
logging.basicConfig(
format="%(message)s",
stream=sys.stdout,
level=getattr(logging, log_level.upper()),
)
def get_logger(name: str) -> structlog.stdlib.BoundLogger:
"""Get logger instance"""
return structlog.get_logger(name)
|