import logging import logging.handlers import sys import time from pathlib import Path from typing import Dict, Any, List, cast, Callable import structlog from structlog.types import FilteringBoundLogger, EventDict from . import __version__ class StructlogFilter(logging.Filter): def __init__(self, logger_names: List[str]): super().__init__() self.logger_names = logger_names self.accept_all = len(logger_names) == 0 def filter(self, record: logging.LogRecord) -> bool: if self.accept_all: return True for logger_name in self.logger_names: if record.name == logger_name or record.name.startswith(logger_name + '.'): return True return False class UvicornStructlogFilter(logging.Filter): def filter(self, record: logging.LogRecord) -> bool: if hasattr(record, 'name') and 'uvicorn.access' in record.name: if hasattr(record, 'getMessage'): msg = record.getMessage() if ' - "' in msg and '" ' in msg: parts = msg.split(' - "') if len(parts) >= 2: client_info = parts[0] request_part = parts[1].split('" ') if len(request_part) >= 2: method_path = request_part[0] status_part = request_part[1] record.client = client_info record.request = method_path record.status = status_part return True def add_timestamp(logger: FilteringBoundLogger, method_name: str, event_dict: EventDict) -> EventDict: event_dict["timestamp"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) return event_dict def add_log_level(logger: FilteringBoundLogger, method_name: str, event_dict: EventDict) -> EventDict: event_dict["level"] = method_name.upper() return event_dict def add_module_info(logger: FilteringBoundLogger, method_name: str, event_dict: EventDict) -> EventDict: if hasattr(logger, '_context') and 'logger_name' in logger._context: logger_name = logger._context['logger_name'] if logger_name.startswith('pyserve'): event_dict["module"] = logger_name elif logger_name.startswith('uvicorn'): event_dict["module"] = 'uvicorn' elif logger_name.startswith('starlette'): event_dict["module"] = 'starlette' else: event_dict["module"] = logger_name return event_dict def filter_module_info(show_module: bool) -> Callable[[FilteringBoundLogger, str, EventDict], EventDict]: def processor(logger: FilteringBoundLogger, method_name: str, event_dict: EventDict) -> EventDict: if not show_module and "module" in event_dict: del event_dict["module"] return event_dict return processor def colored_console_renderer(use_colors: bool = True, show_module: bool = True) -> structlog.dev.ConsoleRenderer: return structlog.dev.ConsoleRenderer( colors=use_colors and hasattr(sys.stderr, 'isatty') and sys.stderr.isatty(), level_styles={ "critical": "\033[35m", # Magenta "error": "\033[31m", # Red "warning": "\033[33m", # Yellow "info": "\033[32m", # Green "debug": "\033[36m", # Cyan }, pad_event=25, ) def plain_console_renderer(show_module: bool = True) -> structlog.dev.ConsoleRenderer: return structlog.dev.ConsoleRenderer( colors=False, pad_event=25, ) def json_renderer() -> structlog.processors.JSONRenderer: return structlog.processors.JSONRenderer(ensure_ascii=False, sort_keys=True) class PyServeLogManager: def __init__(self) -> None: self.configured = False self.handlers: Dict[str, logging.Handler] = {} self.original_handlers: Dict[str, List[logging.Handler]] = {} self._structlog_configured = False def setup_logging(self, config: Dict[str, Any]) -> None: if self.configured: return if 'format' not in config and 'console' not in config and 'files' not in config: level = config.get('level', 'INFO').upper() console_output = config.get('console_output', True) log_file = config.get('log_file', './logs/pyserve.log') config = { 'level': level, 'console_output': console_output, 'format': { 'type': 'standard', 'use_colors': True, 'show_module': True, 'timestamp_format': '%Y-%m-%d %H:%M:%S' }, 'files': [{ 'path': log_file, 'level': level, 'loggers': [], 'max_bytes': 10 * 1024 * 1024, 'backup_count': 5, 'format': { 'type': 'standard', 'use_colors': False, 'show_module': True, 'timestamp_format': '%Y-%m-%d %H:%M:%S' } }] } main_level = config.get('level', 'INFO').upper() console_output = config.get('console_output', True) global_format = config.get('format', {}) console_config = config.get('console', {}) files_config = config.get('files', []) console_format = {**global_format, **console_config.get('format', {})} console_level = console_config.get('level', main_level) self._save_original_handlers() self._clear_all_handlers() self._configure_structlog( main_level=main_level, console_output=console_output, console_format=console_format, console_level=console_level, files_config=files_config ) self._configure_stdlib_loggers(main_level) logger = self.get_logger('pyserve') logger.info( "PyServe logger initialized", version=__version__, level=main_level, console_output=console_output, console_format=console_format.get('type', 'standard') ) for i, file_config in enumerate(files_config): logger.info( "File logging configured", file_index=i, path=file_config.get('path'), level=file_config.get('level', main_level), format_type=file_config.get('format', {}).get('type', 'standard') ) self.configured = True def _configure_structlog( self, main_level: str, console_output: bool, console_format: Dict[str, Any], console_level: str, files_config: List[Dict[str, Any]] ) -> None: shared_processors = [ structlog.stdlib.filter_by_level, add_timestamp, add_log_level, add_module_info, structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, ] if console_output: console_show_module = console_format.get('show_module', True) console_processors = shared_processors.copy() console_processors.append(filter_module_info(console_show_module)) if console_format.get('type') == 'json': console_processors.append(json_renderer()) else: console_processors.append( colored_console_renderer( console_format.get('use_colors', True), console_show_module ) ) console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(getattr(logging, console_level)) console_handler.addFilter(UvicornStructlogFilter()) console_formatter = structlog.stdlib.ProcessorFormatter( processor=colored_console_renderer( console_format.get('use_colors', True), console_show_module ) if console_format.get('type') != 'json' else json_renderer(), ) console_handler.setFormatter(console_formatter) root_logger = logging.getLogger() root_logger.setLevel(logging.DEBUG) root_logger.addHandler(console_handler) self.handlers['console'] = console_handler for i, file_config in enumerate(files_config): file_path = file_config.get('path', './logs/pyserve.log') file_level = file_config.get('level', main_level) file_loggers = file_config.get('loggers', []) max_bytes = file_config.get('max_bytes', 10 * 1024 * 1024) backup_count = file_config.get('backup_count', 5) file_format = file_config.get('format', {}) file_show_module = file_format.get('show_module', True) self._ensure_log_directory(file_path) file_handler = logging.handlers.RotatingFileHandler( file_path, maxBytes=max_bytes, backupCount=backup_count, encoding='utf-8' ) file_handler.setLevel(getattr(logging, file_level)) if file_loggers: file_handler.addFilter(StructlogFilter(file_loggers)) file_processors = shared_processors.copy() file_processors.append(filter_module_info(file_show_module)) file_formatter = structlog.stdlib.ProcessorFormatter( processor=json_renderer() if file_format.get('type') == 'json' else plain_console_renderer(file_show_module), ) file_handler.setFormatter(file_formatter) root_logger = logging.getLogger() root_logger.addHandler(file_handler) self.handlers[f'file_{i}'] = file_handler base_processors = [ structlog.stdlib.filter_by_level, add_timestamp, add_log_level, add_module_info, structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, ] structlog.configure( processors=cast(Any, base_processors + [structlog.stdlib.ProcessorFormatter.wrap_for_formatter]), context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, cache_logger_on_first_use=True, ) self._structlog_configured = True def _configure_stdlib_loggers(self, main_level: str) -> None: library_configs = { 'uvicorn': 'DEBUG' if main_level == 'DEBUG' else 'WARNING', 'uvicorn.access': 'DEBUG' if main_level == 'DEBUG' else 'WARNING', 'uvicorn.error': 'DEBUG' if main_level == 'DEBUG' else 'ERROR', 'uvicorn.asgi': 'DEBUG' if main_level == 'DEBUG' else 'WARNING', 'starlette': 'DEBUG' if main_level == 'DEBUG' else 'WARNING', 'asyncio': 'WARNING', 'concurrent.futures': 'WARNING', 'multiprocessing': 'WARNING', } for logger_name, level in library_configs.items(): logger = logging.getLogger(logger_name) logger.setLevel(getattr(logging, level)) logger.propagate = True def _save_original_handlers(self) -> None: logger_names = ['', 'uvicorn', 'uvicorn.access', 'uvicorn.error', 'starlette'] for name in logger_names: logger = logging.getLogger(name) self.original_handlers[name] = logger.handlers.copy() def _clear_all_handlers(self) -> None: root_logger = logging.getLogger() for handler in root_logger.handlers[:]: root_logger.removeHandler(handler) logger_names = ['uvicorn', 'uvicorn.access', 'uvicorn.error', 'starlette'] for name in logger_names: logger = logging.getLogger(name) for handler in logger.handlers[:]: logger.removeHandler(handler) self.handlers.clear() def _ensure_log_directory(self, log_file: str) -> None: log_dir = Path(log_file).parent log_dir.mkdir(parents=True, exist_ok=True) def get_logger(self, name: str) -> structlog.stdlib.BoundLogger: if not self._structlog_configured: structlog.configure( processors=cast(Any, [ structlog.stdlib.filter_by_level, add_timestamp, add_log_level, structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.stdlib.ProcessorFormatter.wrap_for_formatter, ]), context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, cache_logger_on_first_use=True, ) self._structlog_configured = True return cast(structlog.stdlib.BoundLogger, structlog.get_logger(name).bind(logger_name=name)) def set_level(self, logger_name: str, level: str) -> None: logger = logging.getLogger(logger_name) logger.setLevel(getattr(logging, level.upper())) def add_handler(self, name: str, handler: logging.Handler) -> None: if name not in self.handlers: root_logger = logging.getLogger() root_logger.addHandler(handler) self.handlers[name] = handler def remove_handler(self, name: str) -> None: if name in self.handlers: handler = self.handlers[name] root_logger = logging.getLogger() root_logger.removeHandler(handler) handler.close() del self.handlers[name] def create_access_log( self, method: str, path: str, status_code: int, response_time: float, client_ip: str, user_agent: str = "" ) -> None: access_logger = self.get_logger('pyserve.access') access_logger.info( "HTTP access", method=method, path=path, status_code=status_code, response_time_ms=round(response_time * 1000, 2), client_ip=client_ip, user_agent=user_agent, timestamp_format="access" ) def shutdown(self) -> None: for handler in self.handlers.values(): handler.close() self.handlers.clear() for logger_name, handlers in self.original_handlers.items(): logger = logging.getLogger(logger_name) for handler in handlers: logger.addHandler(handler) self.configured = False self._structlog_configured = False log_manager = PyServeLogManager() def setup_logging(config: Dict[str, Any]) -> None: log_manager.setup_logging(config) def get_logger(name: str) -> structlog.stdlib.BoundLogger: return log_manager.get_logger(name) def create_access_log( method: str, path: str, status_code: int, response_time: float, client_ip: str, user_agent: str = "" ) -> None: log_manager.create_access_log(method, path, status_code, response_time, client_ip, user_agent) def shutdown_logging() -> None: log_manager.shutdown()