Packages
snakepit
0.8.7
0.13.0
0.12.0
0.11.1
0.11.0
0.10.1
0.10.0
0.9.1
0.9.0
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.11
0.6.10
0.6.9
0.6.8
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.1
0.5.0
0.4.3
0.4.2
0.4.1
0.4.0
0.3.3
0.3.2
0.3.1
0.3.0
0.2.1
0.2.0
0.1.2
0.1.1
0.1.0
High-performance pooler and session manager for external language integrations. Supports Python, Node.js, Ruby, and more with gRPC streaming, session management, and production-ready process cleanup.
Current section
Files
Jump to
Current section
Files
priv/python/snakepit_bridge/logging_config.py
"""
Centralized logging configuration for Snakepit Python components.
Environment variables:
SNAKEPIT_LOG_LEVEL: debug, info, warning, error, none (default: error)
"""
from __future__ import annotations
import logging
import os
import sys
from typing import Dict
_LEVELS: Dict[str, int] = {
"debug": logging.DEBUG,
"info": logging.INFO,
"warning": logging.WARNING,
"error": logging.ERROR,
"none": logging.CRITICAL + 1,
}
_DEFAULT_CORRELATION_ID = "-"
class _CorrelationIdFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
if not hasattr(record, "correlation_id") or not record.correlation_id:
record.correlation_id = _DEFAULT_CORRELATION_ID
return True
def _ensure_correlation_filter() -> None:
root_logger = logging.getLogger()
if not any(isinstance(f, _CorrelationIdFilter) for f in root_logger.filters):
root_logger.addFilter(_CorrelationIdFilter())
def configure_logging(force: bool = False) -> None:
"""Configure logging based on environment variables."""
level_str = os.environ.get("SNAKEPIT_LOG_LEVEL", "error").lower()
level = _LEVELS.get(level_str, logging.ERROR)
if level_str == "none":
logging.disable(logging.CRITICAL)
else:
logging.disable(logging.NOTSET)
logging.basicConfig(
level=level,
format=(
"%(asctime)s - [%(threadName)s] - %(name)s - %(levelname)s "
"- [corr=%(correlation_id)s] %(message)s"
),
stream=sys.stderr,
force=force,
)
_ensure_correlation_filter()
# Suppress noisy third-party loggers
logging.getLogger("grpc").setLevel(logging.WARNING)
logging.getLogger("urllib3").setLevel(logging.WARNING)
def get_logger(name: str) -> logging.Logger:
"""Get a logger with the standard Snakepit namespace."""
if name.startswith("snakepit."):
return logging.getLogger(name)
if name.startswith("snakepit_bridge."):
suffix = name[len("snakepit_bridge.") :]
return logging.getLogger(f"snakepit.bridge.{suffix}")
return logging.getLogger(f"snakepit.{name}")