Quart框架第三方任务无法获取app_context的问题求助
我来帮你分析下问题所在,以及对应的解决方案:
问题根源
你遇到的核心问题是Quart的上下文不会自动传递给通过asyncio.loop.create_task创建的任务。虽然你在@app.before_serving里进入了app.app_context(),但第三方库用self.loop.create_task(self.run())创建的任务是直接使用asyncio的原生API,Quart并没有对这类任务做上下文绑定处理——只有通过Quart自身的app.create_task()方法创建的任务,才会自动继承当前的应用上下文。
另外你回调里的代码还有个小笔误:async with app.app_context:应该写成async with app.app_context():,不过这不是主要问题,因为current_app本身在无上下文的环境下就无法被访问。
解决方案1:让第三方任务使用Quart的create_task
如果可以修改第三方库的代码,最简单的方式是让它使用Quart的app.create_task()来创建任务,而不是直接调用asyncio的loop方法:
首先修改ThirdParty类,让它可以接收Quart app实例:
import asyncio class ThirdParty: def __init__(self, app=None, loop=None): self.app = app self.loop = loop or asyncio.get_event_loop() async def start(self): # 优先用Quart的create_task来绑定上下文 if self.app: self.app.create_task(self.run()) else: self.loop.create_task(self.run()) async def run(self): while True: await self.wait_for_trigger() await self.custom_callback_func()
然后在你的主文件里初始化第三方实例时传入app:
from app import create_app from third_party import ThirdParty import asyncio app = create_app() # 初始化时传入Quart app实例 third_party = ThirdParty(app=app) @app.before_serving async def startup(): await third_party.start() @app.after_serving async def shutdown(): await third_party.stop() if __name__ == "__main__": app.run()
最后修改回调函数,不需要手动创建上下文了——因为run任务已经通过app.create_task绑定了上下文:
from quart import current_app as app async def custom_callback_func(): # 直接使用current_app操作数据库即可 db = app.extensions['sqlalchemy'].db # 执行你的数据库更新逻辑
解决方案2:手动传递app实例到回调(无需修改第三方库)
如果无法修改第三方库的代码,你可以直接把Quart的app实例传递给回调函数,避免依赖current_app:
修改回调函数,让它接收app参数:
async def custom_callback_func(app): async with app.app_context(): # 执行数据库更新操作 db = app.extensions['sqlalchemy'].db # ...
然后在startup方法里给第三方库的回调绑定app实例:
from functools import partial @app.before_serving async def startup(): # 把app实例绑定到回调函数上 third_party.custom_callback_func = partial(custom_callback_func, app=app) await third_party.start()
这样回调函数可以直接使用传入的app实例创建上下文,不需要依赖current_app的任务本地上下文。
内容的提问来源于stack exchange,提问作者Eric




