框架开发实战
框架开发实战
开发Python框架需要深入理解软件架构设计原则。本文将介绍如何设计和实现插件系统、中间件、路由和钩子机制。
插件系统设计
插件系统是框架扩展性的关键。一个良好的插件系统应该支持动态加载、版本管理和依赖解析。
import importlib
import os
import sys
from typing import Dict, List, Any, Callable
from abc import ABC, abstractmethod
class PluginManager:
"""插件管理器"""
def __init__(self):
self.plugins: Dict[str, Any] = {}
self.hooks: Dict[str, List[Callable]] = {}
self.plugin_dir = "plugins"
def discover_plugins(self):
"""发现插件"""
if not os.path.exists(self.plugin_dir):
os.makedirs(self.plugin_dir)
for filename in os.listdir(self.plugin_dir):
if filename.endswith('.py') and not filename.startswith('_'):
module_name = filename[:-3]
try:
module = importlib.import_module(f"{self.plugin_dir}.{module_name}")
if hasattr(module, 'Plugin'):
plugin_class = getattr(module, 'Plugin')
plugin = plugin_class()
self.register_plugin(module_name, plugin)
print(f"发现插件: {module_name}")
except Exception as e:
print(f"加载插件 {module_name} 失败: {e}")
def register_plugin(self, name: str, plugin: Any):
"""注册插件"""
self.plugins[name] = plugin
# 注册插件的钩子
if hasattr(plugin, 'get_hooks'):
for hook_name, hook_func in plugin.get_hooks().items():
self.register_hook(hook_name, hook_func)
def register_hook(self, hook_name: str, func: Callable):
"""注册钩子"""
if hook_name not in self.hooks:
self.hooks[hook_name] = []
self.hooks[hook_name].append(func)
def execute_hook(self, hook_name: str, *args, **kwargs):
"""执行钩子"""
results = []
for hook_func in self.hooks.get(hook_name, []):
try:
result = hook_func(*args, **kwargs)
results.append(result)
except Exception as e:
print(f"钩子 {hook_name} 执行失败: {e}")
return results
# 示例插件
class LoggerPlugin:
def get_hooks(self):
return {
'before_request': self.before_request,
'after_request': self.after_request,
}
def before_request(self, request):
print(f"请求开始: {request}")
return request
def after_request(self, response):
print(f"请求结束: {response}")
return response
# 使用插件管理器
plugin_manager = PluginManager()
plugin_manager.register_plugin("logger", LoggerPlugin())
# 测试钩子
print("测试钩子执行：")
plugin_manager.execute_hook("before_request", {"path": "/api/users"})
plugin_manager.execute_hook("after_request", {"status": 200})
中间件模式
中间件是处理请求和响应的组件链。每个中间件可以修改请求、响应或执行其他操作。
from typing import Callable, Any, Dict, List
from dataclasses import dataclass
import time
@dataclass
class Request:
method: str
path: str
headers: Dict[str, str]
body: Any = None
@dataclass
class Response:
status_code: int
headers: Dict[str, str]
body: Any = None
class Middleware:
"""中间件基类"""
def __init__(self, app: Callable):
self.app = app
def __call__(self, request: Request) -> Response:
return self.app(request)
class AuthenticationMiddleware(Middleware):
"""认证中间件"""
def __call__(self, request: Request) -> Response:
# 检查认证头
if 'Authorization' not in request.headers:
return Response(
status_code=401,
headers={'Content-Type': 'application/json'},
body={'error': 'Unauthorized'}
)
# 继续处理
return self.app(request)
class LoggingMiddleware(Middleware):
"""日志中间件"""
def __call__(self, request: Request) -> Response:
start_time = time.time()
print(f"请求开始: {request.method} {request.path}")
response = self.app(request)
duration = time.time() - start_time
print(f"请求结束: {response.status_code} ({duration:.4f}s)")
return response
class CORSMiddleware(Middleware):
"""CORS中间件"""
def __call__(self, request: Request) -> Response:
response = self.app(request)
# 添加CORS头
response.headers['Access-Control-Allow-Origin'] = '*'
response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE'
response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
return response
# 中间件管理器
class MiddlewareChain:
def __init__(self):
self.middlewares: List[Middleware] = []
def add(self, middleware_class: type, **kwargs):
self.middlewares.append((middleware_class, kwargs))
def build(self, app: Callable) -> Callable:
# 从后往前构建中间件链
for middleware_class, kwargs in reversed(self.middlewares):
app = middleware_class(app, **kwargs)
return app
# 使用示例
def basic_app(request: Request) -> Response:
return Response(
status_code=200,
headers={'Content-Type': 'application/json'},
body={'message': 'Hello, World!'}
)
# 构建中间件链
chain = MiddlewareChain()
chain.add(LoggingMiddleware)
chain.add(AuthenticationMiddleware)
chain.add(CORSMiddleware)
app = chain.build(basic_app)
# 测试请求
print("测试中间件链：")
request = Request(
method="GET",
path="/api/users",
headers={"Authorization": "Bearer token123"}
)
response = app(request)
print(f"响应状态: {response.status_code}")
print(f"响应头: {response.headers}")
路由系统
路由系统负责将URL映射到处理函数。一个灵活的路由系统应该支持参数提取、HTTP方法匹配和路由分组。
import re
from typing import Callable, Dict, List, Any, Tuple
from dataclasses import dataclass
@dataclass
class Route:
pattern: str
handler: Callable
methods: List[str]
name: str = None
class Router:
"""路由系统"""
def __init__(self):
self.routes: List[Route] = []
self.url_patterns: List[Tuple[re.Pattern, Route]] = []
def add_route(self, pattern: str, handler: Callable,
methods: List[str] = None, name: str = None):
"""添加路由"""
if methods is None:
methods = ['GET']
route = Route(pattern, handler, methods, name)
self.routes.append(route)
# 编译正则表达式
regex_pattern = self._compile_pattern(pattern)
self.url_patterns.append((regex_pattern, route))
def _compile_pattern(self, pattern: str) -> re.Pattern:
"""编译URL模式"""
# 将URL模式转换为正则表达式
regex = pattern
regex = re.sub(r'<(\w+):(\w+)>', r'(?P<\2>[^/]+)', regex)
regex = re.sub(r'<(\w+)>', r'(?P<\1>[^/]+)', regex)
return re.compile(f'^{regex}$')
def match(self, path: str, method: str = 'GET') -> Tuple[Callable, Dict[str, str]]:
"""匹配路由"""
for regex, route in self.url_patterns:
match = regex.match(path)
if match and method in route.methods:
return route.handler, match.groupdict()
return None, {}
def get(self, pattern: str, handler: Callable, name: str = None):
"""GET路由装饰器"""
def decorator(func):
self.add_route(pattern, func, ['GET'], name)
return func
return decorator
def post(self, pattern: str, handler: Callable, name: str = None):
"""POST路由装饰器"""
def decorator(func):
self.add_route(pattern, func, ['POST'], name)
return func
return decorator
# 使用路由系统
router = Router()
@router.get('/users')
def list_users():
return {'users': ['Alice', 'Bob', 'Charlie']}
@router.get('/users/<int:user_id>')
def get_user(user_id: int):
return {'user_id': user_id, 'name': 'Alice'}
@router.post('/users')
def create_user():
return {'message': 'User created'}
# 测试路由
print("路由测试：")
handler, params = router.match('/users')
if handler:
print(f"匹配路由: /users -> {handler.__name__}")
print(f"结果: {handler()}")
handler, params = router.match('/users/123')
if handler:
print(f"\n匹配路由: /users/123 -> {handler.__name__}")
print(f"参数: {params}")
print(f"结果: {handler(**params)}")
handler, params = router.match('/users', 'POST')
if handler:
print(f"\n匹配路由: POST /users -> {handler.__name__}")
print(f"结果: {handler()}")
钩子机制
钩子机制允许在框架的特定点执行自定义代码。这对于实现事件驱动架构和扩展框架功能非常有用。
from typing import Callable, Dict, List, Any
from dataclasses import dataclass
import time
from functools import wraps
@dataclass
class Hook:
name: str
func: Callable
priority: int = 0
class HookManager:
"""钩子管理器"""
def __init__(self):
self.hooks: Dict[str, List[Hook]] = {}
def register(self, hook_name: str, priority: int = 0):
"""注册钩子装饰器"""
def decorator(func):
hook = Hook(hook_name, func, priority)
if hook_name not in self.hooks:
self.hooks[hook_name] = []
self.hooks[hook_name].append(hook)
self.hooks[hook_name].sort(key=lambda h: h.priority)
return func
return decorator
def execute(self, hook_name: str, *args, **kwargs) -> List[Any]:
"""执行钩子"""
results = []
for hook in self.hooks.get(hook_name, []):
try:
result = hook.func(*args, **kwargs)
results.append(result)
except Exception as e:
print(f"钩子 {hook_name} 执行失败: {e}")
return results
# 使用钩子管理器
hook_manager = HookManager()
# 注册钩子
@hook_manager.register('before_request', priority=1)
def log_request(request):
print(f"日志: 请求 {request}")
return request
@hook_manager.register('before_request', priority=2)
def validate_request(request):
print(f"验证: 请求 {request}")
return request
@hook_manager.register('after_request', priority=1)
def log_response(response):
print(f"日志: 响应 {response}")
return response
# 测试钩子
print("钩子执行测试：")
results = hook_manager.execute('before_request', {'path': '/api'})
print(f"before_request 结果: {results}")
results = hook_manager.execute('after_request', {'status': 200})
print(f"after_request 结果: {results}")
# 事件系统
class EventEmitter:
def __init__(self):
self.listeners: Dict[str, List[Callable]] = {}
def on(self, event: str, func: Callable):
"""注册事件监听器"""
if event not in self.listeners:
self.listeners[event] = []
self.listeners[event].append(func)
def emit(self, event: str, *args, **kwargs):
"""触发事件"""
for func in self.listeners.get(event, []):
func(*args, **kwargs)
# 使用事件系统
emitter = EventEmitter()
def on_user_created(user):
print(f"用户创建: {user}")
def on_user_deleted(user):
print(f"用户删除: {user}")
emitter.on('user_created', on_user_created)
emitter.on('user_deleted', on_user_deleted)
print("\n事件系统测试：")
emitter.emit('user_created', {'id': 1, 'name': 'Alice'})
emitter.emit('user_deleted', {'id': 1, 'name': 'Alice'})
# 完整的框架示例
class SimpleFramework:
def __init__(self):
self.router = Router()
self.middleware_chain = MiddlewareChain()
self.hook_manager = HookManager()
def route(self, pattern: str, methods: List[str] = None):
"""路由装饰器"""
def decorator(func):
self.router.add_route(pattern, func, methods)
return func
return decorator
def use(self, middleware_class: type, **kwargs):
"""添加中间件"""
self.middleware_chain.add(middleware_class, **kwargs)
def before_request(self, func: Callable):
"""请求前钩子"""
self.hook_manager.register('before_request')(func)
return func
def after_request(self, func: Callable):
"""请求后钩子"""
self.hook_manager.register('after_request')(func)
return func
def handle_request(self, request: Request) -> Response:
"""处理请求"""
# 执行请求前钩子
self.hook_manager.execute('before_request', request)
# 构建应用
app = self.router.match
# 应用中间件
app = self.middleware_chain.build(lambda req: Response(200, {}, "OK"))
# 处理请求
response = app(request)
# 执行请求后钩子
self.hook_manager.execute('after_request', response)
return response
# 使用框架
app = SimpleFramework()
@app.route('/api/health')
def health_check():
return {'status': 'healthy'}
@app.before_request
def check_rate_limit(request):
print(f"速率限制检查: {request}")
return request
print("\n框架测试：")
request = Request("GET", "/api/health", {})
response = app.handle_request(request)
print(f"响应: {response}")
框架开发需要综合运用多种设计模式和架构原则。通过实现插件系统、中间件、路由和钩子机制，可以构建出灵活、可扩展的Python框架。掌握这些技术不仅有助于开发自己的框架，还能更好地理解和使用现有的Python框架。