|
|
import logging |
|
|
import sys |
|
|
from typing import Any |
|
|
|
|
|
import structlog |
|
|
|
|
|
|
|
|
def setup_logging(log_level: str = "INFO", log_file: str | None = None) -> None: |
|
|
""" |
|
|
Set up structured logging for the application. |
|
|
|
|
|
Args: |
|
|
log_level (str): The logging level. Defaults to "INFO". |
|
|
log_file (str | None): The path to the log file. If None, logs to stdout. |
|
|
""" |
|
|
logging.basicConfig(level=log_level) |
|
|
|
|
|
processors = [ |
|
|
structlog.contextvars.merge_contextvars, |
|
|
structlog.processors.add_log_level, |
|
|
structlog.processors.StackInfoRenderer(), |
|
|
structlog.processors.TimeStamper(fmt="iso"), |
|
|
structlog.processors.dict_tracebacks, |
|
|
] |
|
|
|
|
|
handler: logging.Handler |
|
|
|
|
|
if log_file: |
|
|
handler = logging.FileHandler(log_file) |
|
|
processors.append(structlog.processors.JSONRenderer()) |
|
|
else: |
|
|
handler = logging.StreamHandler(sys.stdout) |
|
|
processors.append(structlog.dev.ConsoleRenderer()) |
|
|
|
|
|
structlog.configure( |
|
|
processors=processors, |
|
|
logger_factory=structlog.stdlib.LoggerFactory(), |
|
|
wrapper_class=structlog.stdlib.BoundLogger, |
|
|
cache_logger_on_first_use=True, |
|
|
) |
|
|
|
|
|
root_logger = logging.getLogger() |
|
|
root_logger.addHandler(handler) |
|
|
|
|
|
|
|
|
def get_logger(name: str) -> structlog.stdlib.BoundLogger: |
|
|
""" |
|
|
Get a logger instance with the specified name. |
|
|
|
|
|
Args: |
|
|
name (str): The name of the logger. |
|
|
|
|
|
Returns: |
|
|
structlog.stdlib.BoundLogger: The logger instance. |
|
|
""" |
|
|
return structlog.get_logger(name) |
|
|
|
|
|
|
|
|
def bind_extra(logger: structlog.stdlib.BoundLogger, **kwargs: Any) -> structlog.stdlib.BoundLogger: |
|
|
""" |
|
|
Bind extra key-value pairs to the logger. |
|
|
|
|
|
Args: |
|
|
logger (structlog.stdlib.BoundLogger): The logger instance. |
|
|
**kwargs: Key-value pairs to bind to the logger. |
|
|
|
|
|
Returns: |
|
|
structlog.stdlib.BoundLogger: The logger with bound extra information. |
|
|
""" |
|
|
return logger.bind(**kwargs) |
|
|
|