konduktor/pyserve/config.py
Илья Глазунов 537b783726 Add CI/CD pipeline, logging enhancements, and release management
- Create a GitHub Actions workflow for testing with Python 3.12 and 3.13.
- Update Makefile to include release management commands and pipeline checks.
- Document the CI/CD pipeline structure and usage in PIPELINE.md.
- Add structlog for structured logging and enhance logging utilities.
- Implement release management script for automated versioning and tagging.
- Modify logging configuration to support structured logging and improved formatting.
- Update dependencies in pyproject.toml and poetry.lock to include structlog.
- Enhance access logging in server and middleware to include structured data.
2025-09-03 00:13:21 +03:00

321 lines
12 KiB
Python

import yaml
import os
from typing import Dict, Any, List, cast
from dataclasses import dataclass, field
import logging
from .logging_utils import setup_logging
@dataclass
class HttpConfig:
static_dir: str = "./static"
templates_dir: str = "./templates"
@dataclass
class ServerConfig:
host: str = "0.0.0.0"
port: int = 8080
backlog: int = 5
default_root: bool = False
redirect_instructions: Dict[str, str] = field(default_factory=dict)
@dataclass
class SSLConfig:
enabled: bool = False
cert_file: str = "./ssl/cert.pem"
key_file: str = "./ssl/key.pem"
@dataclass
class LogFormatConfig:
type: str = "standard"
use_colors: bool = True
show_module: bool = True
timestamp_format: str = "%Y-%m-%d %H:%M:%S"
@dataclass
class LogFileConfig:
path: str
level: str = "INFO"
format: LogFormatConfig = field(default_factory=LogFormatConfig)
loggers: List[str] = field(default_factory=list) # Список logger'ов для этого файла
max_bytes: int = 10 * 1024 * 1024 # 10MB
backup_count: int = 5
@dataclass
class LogHandlerConfig:
level: str = "INFO"
format: LogFormatConfig = field(default_factory=LogFormatConfig)
@dataclass
class LoggingConfig:
level: str = "INFO"
console_output: bool = True
format: LogFormatConfig = field(default_factory=LogFormatConfig)
console: LogHandlerConfig = field(default_factory=LogHandlerConfig)
files: List[LogFileConfig] = field(default_factory=list)
@dataclass
class RoutingExtensionConfig:
regex_locations: Dict[str, Dict[str, Any]] = field(default_factory=dict)
@dataclass
class ExtensionConfig:
type: str
config: Dict[str, Any] = field(default_factory=dict)
@dataclass
class Config:
http: HttpConfig = field(default_factory=HttpConfig)
server: ServerConfig = field(default_factory=ServerConfig)
ssl: SSLConfig = field(default_factory=SSLConfig)
logging: LoggingConfig = field(default_factory=LoggingConfig)
extensions: List[ExtensionConfig] = field(default_factory=list)
@classmethod
def from_yaml(cls, file_path: str) -> "Config":
try:
with open(file_path, 'r', encoding='utf-8') as f:
data = yaml.safe_load(f)
return cls._from_dict(data)
except FileNotFoundError:
logging.warning(f"Configuration file {file_path} not found. Using default values.")
return cls()
except yaml.YAMLError as e:
logging.error(f"YAML file parsing error {file_path}: {e}")
raise
@classmethod
def _from_dict(cls, data: Dict[str, Any]) -> "Config":
config = cls()
if 'http' in data:
http_data = data['http']
config.http = HttpConfig(
static_dir=http_data.get('static_dir', config.http.static_dir),
templates_dir=http_data.get('templates_dir', config.http.templates_dir)
)
if 'server' in data:
server_data = data['server']
config.server = ServerConfig(
host=server_data.get('host', config.server.host),
port=server_data.get('port', config.server.port),
backlog=server_data.get('backlog', config.server.backlog),
default_root=server_data.get('default_root', config.server.default_root),
redirect_instructions=server_data.get('redirect_instructions', {})
)
if 'ssl' in data:
ssl_data = data['ssl']
config.ssl = SSLConfig(
enabled=ssl_data.get('enabled', config.ssl.enabled),
cert_file=ssl_data.get('cert_file', config.ssl.cert_file),
key_file=ssl_data.get('key_file', config.ssl.key_file)
)
if 'logging' in data:
log_data = data['logging']
format_data = log_data.get('format', {})
global_format = LogFormatConfig(
type=format_data.get('type', 'standard'),
use_colors=format_data.get('use_colors', True),
show_module=format_data.get('show_module', True),
timestamp_format=format_data.get('timestamp_format', '%Y-%m-%d %H:%M:%S')
)
console_data = log_data.get('console', {})
console_format_data = console_data.get('format', {})
console_format = LogFormatConfig(
type=console_format_data.get('type', global_format.type),
use_colors=console_format_data.get('use_colors', global_format.use_colors),
show_module=console_format_data.get('show_module', global_format.show_module),
timestamp_format=console_format_data.get('timestamp_format', global_format.timestamp_format)
)
console_config = LogHandlerConfig(
level=console_data.get('level', log_data.get('level', 'INFO')),
format=console_format
)
files_config = []
if 'log_file' in log_data:
default_file_format = LogFormatConfig(
type=global_format.type,
use_colors=False,
show_module=global_format.show_module,
timestamp_format=global_format.timestamp_format
)
default_file = LogFileConfig(
path=log_data['log_file'],
level=log_data.get('level', 'INFO'),
format=default_file_format,
loggers=[], # Empty list means including all loggers
max_bytes=10 * 1024 * 1024,
backup_count=5
)
files_config.append(default_file)
if 'files' in log_data:
for file_data in log_data['files']:
file_format_data = file_data.get('format', {})
file_format = LogFormatConfig(
type=file_format_data.get('type', global_format.type),
use_colors=file_format_data.get('use_colors', False),
show_module=file_format_data.get('show_module', global_format.show_module),
timestamp_format=file_format_data.get('timestamp_format', global_format.timestamp_format)
)
file_config = LogFileConfig(
path=file_data.get('path', './logs/pyserve.log'),
level=file_data.get('level', log_data.get('level', 'INFO')),
format=file_format,
loggers=file_data.get('loggers', []),
max_bytes=file_data.get('max_bytes', 10 * 1024 * 1024),
backup_count=file_data.get('backup_count', 5)
)
files_config.append(file_config)
if 'show_module' in console_format_data:
print(
"\033[33mWARNING: Parameter 'show_module' in console.format in development and may work incorrectly\033[0m"
)
console_config.format.show_module = console_format_data.get('show_module')
for i, file_data in enumerate(log_data.get('files', [])):
if 'format' in file_data and 'show_module' in file_data['format']:
print(
f"\033[33mWARNING: Parameter 'show_module' in files[{i}].format in development and may work incorrectly\033[0m"
)
if not files_config:
default_file_format = LogFormatConfig(
type=global_format.type,
use_colors=False,
show_module=global_format.show_module,
timestamp_format=global_format.timestamp_format
)
default_file = LogFileConfig(
path='./logs/pyserve.log',
level=log_data.get('level', 'INFO'),
format=default_file_format,
loggers=[],
max_bytes=10 * 1024 * 1024,
backup_count=5
)
files_config.append(default_file)
config.logging = LoggingConfig(
level=log_data.get('level', 'INFO'),
console_output=log_data.get('console_output', True),
format=global_format,
console=console_config,
files=files_config
)
if 'extensions' in data:
for ext_data in data['extensions']:
extension = ExtensionConfig(
type=ext_data.get('type', ''),
config=ext_data.get('config', {})
)
config.extensions.append(extension)
return config
def validate(self) -> bool:
errors = []
if not os.path.exists(self.http.static_dir):
errors.append(f"Static directory does not exist: {self.http.static_dir}")
if self.ssl.enabled:
if not os.path.exists(self.ssl.cert_file):
errors.append(f"SSL certificate not found: {self.ssl.cert_file}")
if not os.path.exists(self.ssl.key_file):
errors.append(f"SSL key not found: {self.ssl.key_file}")
if not (1 <= self.server.port <= 65535):
errors.append(f"Invalid port: {self.server.port}")
valid_log_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
if self.logging.level.upper() not in valid_log_levels:
errors.append(f"Invalid logging level: {self.logging.level}")
if self.logging.console.level.upper() not in valid_log_levels:
errors.append(f"Invalid console logging level: {self.logging.console.level}")
valid_format_types = ['standard', 'json']
if self.logging.format.type not in valid_format_types:
errors.append(f"Invalid logging format type: {self.logging.format.type}")
if self.logging.console.format.type not in valid_format_types:
errors.append(f"Invalid console format type: {self.logging.console.format.type}")
for i, file_config in enumerate(self.logging.files):
if file_config.level.upper() not in valid_log_levels:
errors.append(f"Invalid file[{i}] logging level: {file_config.level}")
if file_config.format.type not in valid_format_types:
errors.append(f"Invalid file[{i}] format type: {file_config.format.type}")
log_dir = os.path.dirname(file_config.path)
if log_dir and not os.path.exists(log_dir):
try:
os.makedirs(log_dir, exist_ok=True)
except OSError as e:
errors.append(f"Unable to create log directory for {file_config.path}: {e}")
if errors:
for error in errors:
logging.error(f"Configuration error: {error}")
return False
return True
def setup_logging(self) -> None:
config_dict = {
'level': self.logging.level,
'console_output': self.logging.console_output,
'format': {
'type': self.logging.format.type,
'use_colors': self.logging.format.use_colors,
'show_module': self.logging.format.show_module,
'timestamp_format': self.logging.format.timestamp_format
},
'console': {
'level': self.logging.console.level,
'format': {
'type': self.logging.console.format.type,
'use_colors': self.logging.console.format.use_colors,
'show_module': self.logging.console.format.show_module,
'timestamp_format': self.logging.console.format.timestamp_format
}
},
'files': []
}
for file_config in self.logging.files:
file_dict = {
'path': file_config.path,
'level': file_config.level,
'loggers': file_config.loggers,
'max_bytes': file_config.max_bytes,
'backup_count': file_config.backup_count,
'format': {
'type': file_config.format.type,
'use_colors': file_config.format.use_colors,
'show_module': file_config.format.show_module,
'timestamp_format': file_config.format.timestamp_format
}
}
cast(List[Dict[str, Any]], config_dict['files']).append(file_dict)
setup_logging(config_dict)