You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

Quart框架第三方任务无法获取app_context的问题求助

解决Quart第三方任务中"Attempt to access app outside of a relevant 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

火山引擎 最新活动