# 架构设计原则对照表
principles = {'SOLID': {'SRP': '单一职责原则','OCP': '开闭原则','LSP': '里氏替换原则','ISP': '接口隔离原则','DIP': '依赖倒置原则'},'CORE': {'模块化': '高内聚低耦合','可配置': '约定优于配置','可扩展': '插件化架构','可观测': '完善的日志和监控','可测试': '依赖注入支持'}
}

# Python包常用架构模式
architecture_patterns = {'分层架构': {'示例': '业务逻辑层/数据访问层/表示层','适用': '复杂业务系统'},'插件架构': {'示例': '核心系统+插件注册机制','适用': '需要扩展性的框架'},'微内核': {'示例': '核心引擎+可替换组件','适用': 'IDE/编辑器类应用'},'事件驱动': {'示例': '事件总线+处理器','适用': '异步处理系统'}
}

from abc import ABC, abstractmethod
from typing import Dict, Anyclass DataProcessor(ABC):"""数据处理抽象基类"""@abstractmethoddef initialize(self, config: Dict[str, Any]) -> None:"""初始化处理器"""pass@abstractmethoddef process(self, data: Any) -> Any:"""处理数据并返回结果"""pass@abstractmethoddef shutdown(self) -> None:"""清理资源"""passclass AnalysisEngine:"""分析引擎核心类"""def __init__(self):self._processors = {}def register_processor(self, name: str, processor: DataProcessor) -> None:"""注册数据处理器"""if not isinstance(processor, DataProcessor):raise TypeError("Must be DataProcessor instance")self._processors[name] = processordef analyze(self, data: Any, processor_name: str) -> Any:"""使用指定处理器分析数据"""processor = self._processors.get(processor_name)if not processor:raise ValueError(f"Unknown processor: {processor_name}")return processor.process(data)

from dataclasses import dataclass
from dependency_injector import containers, providers@dataclass
class DatabaseConfig:host: strport: intuser: strpassword: strclass DatabaseService:def __init__(self, config: DatabaseConfig):self.config = configdef query(self, sql: str):# 实际数据库操作passclass CoreContainer(containers.DeclarativeContainer):"""依赖注入容器"""config = providers.Configuration()db_config = providers.Singleton(DatabaseConfig,host=config.db.host,port=config.db.port,user=config.db.user,password=config.db.password)database = providers.Singleton(DatabaseService,config=db_config)# 使用示例
container = CoreContainer()
container.config.from_dict({"db": {"host": "localhost","port": 5432,"user": "admin","password": "secret"}
})db_service = container.database()
result = db_service.query("SELECT * FROM users")

import importlib
import pkgutil
from pathlib import Pathclass PluginManager:"""插件管理器"""def __init__(self, plugin_dir: str = "plugins"):self.plugin_dir = Path(plugin_dir)self._plugins = {}def discover(self) -> None:"""发现并加载所有插件"""if not self.plugin_dir.exists():self.plugin_dir.mkdir()for finder, name, _ in pkgutil.iter_modules([str(self.plugin_dir)]):module = finder.find_module(name).load_module(name)if hasattr(module, 'register'):module.register(self)def register(self, name: str, plugin: Any) -> None:"""注册插件"""self._plugins[name] = plugindef get_plugin(self, name: str) -> Any:"""获取插件实例"""return self._plugins.get(name)# 插件示例 (plugins/example_plugin.py)
"""
def register(manager):manager.register('example', ExamplePlugin())
"""

[project.entry-points."my_package.plugins"]
example = "example_plugin:register"
csv = "csv_plugin:register"
json = "json_plugin:register"

from importlib.metadata import entry_pointsdef load_entry_points():plugins = {}for entry_point in entry_points().get('my_package.plugins', []):register_func = entry_point.load()plugins[entry_point.name] = register_funcreturn plugins

from typing import Dict, Any
import json
import os
from pathlib import Pathclass ConfigManager:"""分层配置管理器"""def __init__(self):self._config = {'defaults': {},'user': {},'environment': {},'runtime': {}}def load_defaults(self, config_dict: Dict[str, Any]) -> None:"""加载默认配置"""self._config['defaults'].update(config_dict)def load_user_config(self, filepath: str) -> None:"""加载用户配置文件"""path = Path(filepath)if path.exists():with open(path) as f:self._config['user'].update(json.load(f))def load_environment(self) -> None:"""加载环境变量配置"""self._config['environment'].update({key.lower(): os.getenv(key)for key in os.environif key.startswith('MY_PKG_')})def get(self, key: str, default: Any = None) -> Any:"""获取配置值（按优先级）"""for layer in ['runtime', 'environment', 'user', 'defaults']:if key in self._config[layer]:return self._config[layer][key]return default

from pydantic import BaseModel, validator
from typing import Literalclass DatabaseConfig(BaseModel):"""数据库配置模型"""engine: Literal['mysql', 'postgresql', 'sqlite']host: str = 'localhost'port: int = 3306timeout: int = 30@validator('port')def validate_port(cls, v):if not 0 < v < 65535:raise ValueError('Port must be between 1-65534')return v# 使用示例
try:config = DatabaseConfig(**user_config)
except ValueError as e:print(f"Invalid config: {e}")

import logging
import json
from typing import Dict, Anyclass JSONFormatter(logging.Formatter):"""JSON日志格式化器"""def format(self, record: logging.LogRecord) -> str:log_data = {'timestamp': self.formatTime(record),'level': record.levelname,'message': record.getMessage(),'module': record.module,'function': record.funcName,'line': record.lineno}if hasattr(record, 'context'):log_data.update(record.context)return json.dumps(log_data)def setup_logging():"""配置结构化日志"""logger = logging.getLogger('my_pkg')logger.setLevel(logging.INFO)handler = logging.StreamHandler()handler.setFormatter(JSONFormatter())logger.addHandler(handler)return logger# 使用示例
logger = setup_logging()
logger.info("Processing started", extra={'context': {'job_id': 123,'input_size': 1024}
})

from prometheus_client import Counter, Gauge, start_http_serverclass Metrics:"""监控指标收集器"""def __init__(self):self.requests = Counter('app_requests_total','Total number of requests',['endpoint', 'method'])self.errors = Counter('app_errors_total','Total number of errors',['type'])self.processing_time = Gauge('app_processing_seconds','Time spent processing requests')# 使用示例
metrics = Metrics()
start_http_server(8000)@contextmanager
def track_processing():start = time.time()try:yieldexcept Exception as e:metrics.errors.labels(type=type(e).__name__).inc()raisefinally:metrics.processing_time.set(time.time() - start)

class PackageError(Exception):"""基础异常类"""code = 1000def __init__(self, message: str, context: Dict = None):super().__init__(message)self.context = context or {}class ConfigurationError(PackageError):"""配置异常"""code = 1001class ProcessingError(PackageError):"""处理异常"""code = 2000class PluginError(PackageError):"""插件异常"""code = 3000def handle_error(func):"""统一异常处理装饰器"""def wrapper(*args, **kwargs):try:return func(*args, **kwargs)except PackageError as e:logger.error(f"Error[{e.code}]: {str(e)}", extra=e.context)raiseexcept Exception as e:logger.exception("Unexpected error occurred")raise PackageError("Internal error") from ereturn wrapper

from enum import IntEnumclass ErrorCode(IntEnum):"""错误代码枚举"""# 配置错误 (1000-1999)INVALID_CONFIG = 1001MISSING_REQUIRED = 1002# 处理错误 (2000-2999)TIMEOUT = 2001INVALID_INPUT = 2002# 插件错误 (3000-3999)PLUGIN_LOAD_FAILED = 3001PLUGIN_INTERFACE_ERROR = 3002ERROR_MESSAGES = {ErrorCode.INVALID_CONFIG: "Invalid configuration: {details}",ErrorCode.MISSING_REQUIRED: "Missing required parameter: {param}",ErrorCode.TIMEOUT: "Operation timed out after {seconds}s",ErrorCode.INVALID_INPUT: "Invalid input data: {reason}",ErrorCode.PLUGIN_LOAD_FAILED: "Failed to load plugin: {plugin}",ErrorCode.PLUGIN_INTERFACE_ERROR: "Plugin interface error: {method}"
}def raise_error(code: ErrorCode, **kwargs):"""根据错误码抛出异常"""message = ERROR_MESSAGES[code].format(**kwargs)raise PackageError(message, {'error_code': code.value, **kwargs})

from abc import ABC, abstractmethod
from typing import BinaryIO, Optionalclass StorageBackend(ABC):"""抽象存储接口"""@abstractmethoddef save(self, key: str, data: BinaryIO) -> None:"""保存数据"""pass@abstractmethoddef load(self, key: str) -> Optional[BinaryIO]:"""加载数据"""pass@abstractmethoddef delete(self, key: str) -> bool:"""删除数据"""passclass S3Storage(StorageBackend):"""S3存储实现"""def __init__(self, bucket_name: str):import boto3self.client = boto3.client('s3')self.bucket = bucket_namedef save(self, key: str, data: BinaryIO) -> None:self.client.upload_fileobj(data, self.bucket, key)def load(self, key: str) -> Optional[BinaryIO]:# 实现略passdef delete(self, key: str) -> bool:# 实现略passclass StorageManager:"""存储管理器"""def __init__(self, backend: StorageBackend):self.backend = backenddef store_data(self, key: str, data: bytes) -> None:import iowith io.BytesIO(data) as buffer:self.backend.save(key, buffer)

from typing import Protocol, runtime_checkable@runtime_checkable
class AnalyzerProtocol(Protocol):"""分析器协议"""def analyze(self, text: str) -> dict:...@propertydef version(self) -> str:...class LegacyAnalyzer:"""遗留分析器适配"""def process(self, content: str) -> dict:return {"result": len(content)}def get_version(self) -> str:return "1.0"class AnalyzerAdapter:"""适配器使遗留分析器符合协议"""def __init__(self, legacy_analyzer: LegacyAnalyzer):self.analyzer = legacy_analyzerdef analyze(self, text: str) -> dict:return self.analyzer.process(text)@propertydef version(self) -> str:return self.analyzer.get_version()# 使用示例
legacy = LegacyAnalyzer()
adapter = AnalyzerAdapter(legacy)if isinstance(adapter, AnalyzerProtocol):print("Adapter conforms to protocol")

import arrayclass HighPerformanceProcessor:"""高性能二进制处理器"""def __init__(self, data: bytes):self.data = memoryview(data)def find_pattern(self, pattern: bytes) -> int:"""使用内存视图高效查找模式"""view = self.datapattern_view = memoryview(pattern)pattern_len = len(pattern_view)for i in range(len(view) - pattern_len + 1):if view[i:i+pattern_len] == pattern_view:return ireturn -1# 使用示例
processor = HighPerformanceProcessor(b"some binary data\x00\x01\x02")
position = processor.find_pattern(b"\x00\x01")

# distutils: language_level=3
import arraycdef class FastProcessor:cdef unsigned char[:] data_viewdef __cinit__(self, data):self.data_view = datacpdef int find_pattern(self, bytes pattern):cdef unsigned char[:] pattern_view = patterncdef int pattern_len = len(pattern_view)cdef int data_len = len(self.data_view)for i in range(data_len - pattern_len + 1):match = Truefor j in range(pattern_len):if self.data_view[i + j] != pattern_view[j]:match = Falsebreakif match:return ireturn -1