import asyncio import logging import sys from DbKit.Db import Db # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def patch(): # 在Windows上使用SelectorEventLoopPolicy可以减少SSL错误 if sys.platform == 'win32': asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) # 为Db类添加临时的异步上下文管理器支持 def patch_db_context_manager(): # 为Db类添加异步上下文管理器支持 if not hasattr(Db, '__aenter__'): async def __aenter__(self): return self async def __aexit__(self, exc_type, exc_val, exc_tb): # 简单的清理,不做实际操作,避免事件循环问题 return False # 动态添加方法 Db.__aenter__ = __aenter__ Db.__aexit__ = __aexit__ # 应用补丁 patch_db_context_manager()