92 lines
2.5 KiB
Python
92 lines
2.5 KiB
Python
"""
|
|
StreamFlow Logging Configuration
|
|
Structured logging with JSON support for production
|
|
"""
|
|
import logging
|
|
import logging.config
|
|
import sys
|
|
from typing import Optional
|
|
|
|
|
|
def setup_logging(debug: bool = False, json_format: bool = False) -> logging.Logger:
|
|
"""
|
|
Configure application logging.
|
|
|
|
Args:
|
|
debug: Enable debug level logging
|
|
json_format: Use JSON format for production log aggregation
|
|
|
|
Returns:
|
|
Root logger instance
|
|
"""
|
|
log_level = "DEBUG" if debug else "INFO"
|
|
|
|
if json_format:
|
|
# JSON format for production (compatible with log aggregators)
|
|
log_format = '{"time": "%(asctime)s", "level": "%(levelname)s", "name": "%(name)s", "message": "%(message)s"}'
|
|
else:
|
|
# Human-readable format for development
|
|
log_format = "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
|
|
|
|
config = {
|
|
"version": 1,
|
|
"disable_existing_loggers": False,
|
|
"formatters": {
|
|
"standard": {
|
|
"format": log_format,
|
|
"datefmt": "%Y-%m-%d %H:%M:%S"
|
|
}
|
|
},
|
|
"handlers": {
|
|
"console": {
|
|
"class": "logging.StreamHandler",
|
|
"level": log_level,
|
|
"formatter": "standard",
|
|
"stream": "ext://sys.stdout"
|
|
}
|
|
},
|
|
"loggers": {
|
|
"streamflow": {
|
|
"level": log_level,
|
|
"handlers": ["console"],
|
|
"propagate": False
|
|
},
|
|
"uvicorn": {
|
|
"level": "INFO",
|
|
"handlers": ["console"],
|
|
"propagate": False
|
|
},
|
|
"uvicorn.access": {
|
|
"level": "INFO",
|
|
"handlers": ["console"],
|
|
"propagate": False
|
|
}
|
|
},
|
|
"root": {
|
|
"level": log_level,
|
|
"handlers": ["console"]
|
|
}
|
|
}
|
|
|
|
logging.config.dictConfig(config)
|
|
return logging.getLogger("streamflow")
|
|
|
|
|
|
def get_logger(name: Optional[str] = None) -> logging.Logger:
|
|
"""
|
|
Get a logger instance for a module.
|
|
|
|
Args:
|
|
name: Module name (e.g., __name__). If None, returns root streamflow logger.
|
|
|
|
Returns:
|
|
Logger instance
|
|
|
|
Example:
|
|
from logging_config import get_logger
|
|
logger = get_logger(__name__)
|
|
logger.info("Processing request", extra={"slug": slug})
|
|
"""
|
|
if name:
|
|
return logging.getLogger(f"streamflow.{name}")
|
|
return logging.getLogger("streamflow")
|