| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162 |
- """应用生命周期管理 —— 回调注册模式"""
- import asyncio
- import logging
- from typing import Callable, Awaitable, List
- logger = logging.getLogger(__name__)
- StartupCallback = Callable[[], Awaitable[None]]
- ShutdownCallback = Callable[[], Awaitable[None]]
- class AppContext:
- """管理应用启动/关闭生命周期
- 使用方式：
- ctx = AppContext()
- ctx.on_startup(db.init_pools)
- ctx.on_startup(log_service.start)
- ctx.on_shutdown(db.close_pools)
- ctx.on_shutdown(log_service.stop)
- # 在 Quart/FastAPI 生命周期中调用
- await ctx.start_up()
- await ctx.shutdown()
- """
- def __init__(self):
- self._startup_callbacks: List[StartupCallback] = []
- self._shutdown_callbacks: List[ShutdownCallback] = []
- def on_startup(self, callback: StartupCallback) -> "AppContext":
- """注册启动回调（支持链式调用）"""
- self._startup_callbacks.append(callback)
- return self
- def on_shutdown(self, callback: ShutdownCallback) -> "AppContext":
- """注册关闭回调（支持链式调用）"""
- self._shutdown_callbacks.append(callback)
- return self
- async def start_up(self) -> None:
- """按注册顺序执行所有启动回调"""
- logger.info("应用启动中...")
- results = await asyncio.gather(
- *(cb() for cb in self._startup_callbacks),
- return_exceptions=True,
- )
- for i, result in enumerate(results):
- if isinstance(result, Exception):
- logger.error(f"启动回调 [{i}] 失败: {result}")
- logger.info("应用启动完成")
- async def shutdown(self) -> None:
- """按注册逆序执行所有关闭回调"""
- logger.info("应用关闭中...")
- for callback in reversed(self._shutdown_callbacks):
- try:
- await callback()
- except Exception as e:
- logger.error(f"关闭回调失败: {e}")
- logger.info("应用已关闭")
|
