import logging import logging.handlers import sys import time import json from pathlib import Path from typing import Dict, Any, List from . import __version__ class LoggerFilter(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 UvicornLogFilter(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.msg = f"Access: {client_info} - {method_path} - {status_part}" return True class PyServeFormatter(logging.Formatter): COLORS = { 'DEBUG': '\033[36m', # Cyan 'INFO': '\033[32m', # Green 'WARNING': '\033[33m', # Yellow 'ERROR': '\033[31m', # Red 'CRITICAL': '\033[35m', # Magenta 'RESET': '\033[0m' # Reset } def __init__(self, use_colors: bool = True, show_module: bool = True, timestamp_format: str = "%Y-%m-%d %H:%M:%S", *args: Any, **kwargs: Any): super().__init__(*args, **kwargs) self.use_colors = use_colors and hasattr(sys.stderr, 'isatty') and sys.stderr.isatty() self.show_module = show_module self.timestamp_format = timestamp_format def format(self, record: logging.LogRecord) -> str: if self.use_colors: levelname = record.levelname if levelname in self.COLORS: record.levelname = f"{self.COLORS[levelname]}{levelname}{self.COLORS['RESET']}" if self.show_module and hasattr(record, 'name'): name = record.name if name.startswith('uvicorn'): record.name = 'uvicorn' elif name.startswith('pyserve'): pass elif name.startswith('starlette'): record.name = 'starlette' return super().format(record) class PyServeJSONFormatter(logging.Formatter): def __init__(self, timestamp_format: str = "%Y-%m-%d %H:%M:%S", *args: Any, **kwargs: Any): super().__init__(*args, **kwargs) self.timestamp_format = timestamp_format def format(self, record: logging.LogRecord) -> str: log_entry = { 'timestamp': time.strftime(self.timestamp_format, time.localtime(record.created)), 'level': record.levelname, 'logger': record.name, 'message': record.getMessage(), 'module': record.module, 'function': record.funcName, 'line': record.lineno, 'thread': record.thread, 'thread_name': record.threadName, } if record.exc_info: log_entry['exception'] = self.formatException(record.exc_info) for key, value in record.__dict__.items(): if key not in ['name', 'msg', 'args', 'levelname', 'levelno', 'pathname', 'filename', 'module', 'lineno', 'funcName', 'created', 'msecs', 'relativeCreated', 'thread', 'threadName', 'processName', 'process', 'getMessage', 'exc_info', 'exc_text', 'stack_info']: log_entry[key] = value return json.dumps(log_entry, ensure_ascii=False, default=str) class AccessLogHandler(logging.Handler): def __init__(self, logger_name: str = 'pyserve.access'): super().__init__() self.access_logger = logging.getLogger(logger_name) def emit(self, record: logging.LogRecord) -> None: self.access_logger.handle(record) class PyServeLogManager: def __init__(self) -> None: self.configured = False self.handlers: Dict[str, logging.Handler] = {} self.loggers: Dict[str, logging.Logger] = {} self.original_handlers: Dict[str, List[logging.Handler]] = {} def _create_formatter(self, format_config: Dict[str, Any]) -> logging.Formatter: format_type = format_config.get('type', 'standard').lower() use_colors = format_config.get('use_colors', True) show_module = format_config.get('show_module', True) timestamp_format = format_config.get('timestamp_format', '%Y-%m-%d %H:%M:%S') if format_type == 'json': return PyServeJSONFormatter(timestamp_format=timestamp_format) else: if format_type == 'json': fmt = None else: fmt = '%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s' return PyServeFormatter( use_colors=use_colors, show_module=show_module, timestamp_format=timestamp_format, fmt=fmt ) 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() root_logger = logging.getLogger() root_logger.setLevel(logging.DEBUG) if console_output: console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(getattr(logging, console_level)) if console_format.get('type') == 'json': console_formatter = self._create_formatter(console_format) else: console_formatter = PyServeFormatter( use_colors=console_format.get('use_colors', True), show_module=console_format.get('show_module', True), timestamp_format=console_format.get('timestamp_format', '%Y-%m-%d %H:%M:%S'), fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) console_handler.setFormatter(console_formatter) console_handler.addFilter(UvicornLogFilter()) 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 = {**global_format, **file_config.get('format', {})} 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_format.get('type') == 'json': file_formatter = self._create_formatter(file_format) else: file_formatter = PyServeFormatter( use_colors=file_format.get('use_colors', False), show_module=file_format.get('show_module', True), timestamp_format=file_format.get('timestamp_format', '%Y-%m-%d %H:%M:%S'), fmt='%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s' ) file_handler.setFormatter(file_formatter) file_handler.addFilter(UvicornLogFilter()) if file_loggers: logger_filter = LoggerFilter(file_loggers) file_handler.addFilter(logger_filter) root_logger.addHandler(file_handler) self.handlers[f'file_{i}'] = file_handler self._configure_library_loggers(main_level) self._intercept_uvicorn_logging() pyserve_logger = logging.getLogger('pyserve') pyserve_logger.setLevel(getattr(logging, main_level)) self.loggers['pyserve'] = pyserve_logger pyserve_logger.info(f"PyServe v{__version__} - Logger initialized") pyserve_logger.info(f"Logging level: {main_level}") pyserve_logger.info(f"Console output: {'enabled' if console_output else 'disabled'}") pyserve_logger.info(f"Console format: {console_format.get('type', 'standard')}") for i, file_config in enumerate(files_config): file_path = file_config.get('path', './logs/pyserve.log') file_loggers = file_config.get('loggers', []) file_format = file_config.get('format', {}) pyserve_logger.info(f"Log file[{i}]: {file_path}") pyserve_logger.info(f"File format[{i}]: {file_format.get('type', 'standard')}") if file_loggers: pyserve_logger.info(f"File loggers[{i}]: {', '.join(file_loggers)}") else: pyserve_logger.info(f"File loggers[{i}]: all loggers") self.configured = 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) handler.close() 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) handler.close() 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 _configure_library_loggers(self, main_level: str) -> None: library_configs = { # Uvicorn and related - only in DEBUG mode '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 - only in DEBUG mode 'starlette': 'DEBUG' if main_level == 'DEBUG' else 'WARNING', 'asyncio': 'WARNING', 'concurrent.futures': 'WARNING', 'multiprocessing': 'WARNING', 'pyserve': main_level, 'pyserve.server': main_level, 'pyserve.routing': main_level, 'pyserve.extensions': main_level, 'pyserve.config': main_level, } for logger_name, level in library_configs.items(): logger = logging.getLogger(logger_name) logger.setLevel(getattr(logging, level)) if logger_name.startswith('uvicorn') and logger_name != 'uvicorn': logger.propagate = False self.loggers[logger_name] = logger def _intercept_uvicorn_logging(self) -> None: uvicorn_logger = logging.getLogger('uvicorn') uvicorn_access_logger = logging.getLogger('uvicorn.access') for handler in uvicorn_logger.handlers[:]: uvicorn_logger.removeHandler(handler) for handler in uvicorn_access_logger.handlers[:]: uvicorn_access_logger.removeHandler(handler) uvicorn_logger.propagate = True uvicorn_access_logger.propagate = True def get_logger(self, name: str) -> logging.Logger: if name not in self.loggers: logger = logging.getLogger(name) self.loggers[name] = logger return self.loggers[name] def set_level(self, logger_name: str, level: str) -> None: if logger_name in self.loggers: self.loggers[logger_name].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: root_logger = logging.getLogger() root_logger.removeHandler(self.handlers[name]) self.handlers[name].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') log_message = f'{client_ip} - - [{time.strftime("%d/%b/%Y:%H:%M:%S %z")}] ' \ f'"{method} {path} HTTP/1.1" {status_code} - ' \ f'"{user_agent}" {response_time:.3f}s' access_logger.info(log_message) 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.loggers.clear() self.configured = False log_manager = PyServeLogManager() def setup_logging(config: Dict[str, Any]) -> None: log_manager.setup_logging(config) def get_logger(name: str) -> logging.Logger: 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()