Quart @stream_with_context装饰器无法持久化g全局对象问题求助

问题现象 #

原本期望Quart的@stream_with_context装饰器能让请求上下文的g对象在异步生成器中可用，但实际运行时，在生成器内访问g.username会抛出如下错误：

raise AttributeError(name) from None
AttributeError: username

测试代码如下：

import asyncio
from datetime import datetime

from quart import Blueprint, stream_with_context, g
from quart import Quart
from quart.views import MethodView

blueprint = Blueprint('main', __name__)


class StreamTestResource(MethodView):
    async def get(self):
        g.username = 'test'

        @stream_with_context
        async def async_generator():
            print(g.username) # <----- 此行抛出错误
            for i in range(5):
                await asyncio.sleep(1)
                time = datetime.isoformat(datetime.now())
                yield f'data: {time}\n\n'

        return async_generator(), 200, {
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Transfer-Encoding': 'chunked',
        }


blueprint.add_url_rule('/ping', view_func=StreamTestResource.as_view('ping'))

app = Quart(__name__)
app.register_blueprint(blueprint)

if __name__ == '__main__':
    app.run()

原因分析 #

Quart的g对象是与请求上下文绑定的临时存储，当请求处理方法（如get）执行完毕后，对应的请求上下文会被销毁。异步生成器是在方法返回后才开始迭代的，此时原上下文已丢失，导致无法访问之前设置的g.username属性。stream_with_context本身用于保留流式响应的上下文，但无法阻止请求方法结束后原上下文的销毁。

解决方案 #

方式一：提前捕获属性到局部变量（推荐） #

直接将g中需要的属性赋值给局部变量，在生成器内使用该变量，完全脱离对请求上下文的依赖，这是最可靠的方式：

import asyncio
from datetime import datetime

from quart import Blueprint, stream_with_context, g
from quart import Quart, Response
from quart.views import MethodView

blueprint = Blueprint('main', __name__)


class StreamTestResource(MethodView):
    async def get(self):
        g.username = 'test'
        # 提前捕获属性到局部变量
        current_username = g.username

        @stream_with_context
        async def async_generator():
            print(current_username)
            for i in range(5):
                await asyncio.sleep(1)
                time = datetime.isoformat(datetime.now())
                yield f'data: {time}\n\n'

        return Response(
            async_generator(),
            200,
            headers={
                'Content-Type': 'text/event-stream',
                'Cache-Control': 'no-cache',
                'Transfer-Encoding': 'chunked',
            }
        )


blueprint.add_url_rule('/ping', view_func=StreamTestResource.as_view('ping'))

app = Quart(__name__)
app.register_blueprint(blueprint)

if __name__ == '__main__':
    app.run()

方式二：复制请求上下文到生成器 #

如果必须依赖g对象，可以使用copy_current_request_context装饰器复制当前请求上下文，确保生成器迭代时上下文有效：

import asyncio
from datetime import datetime

from quart import Blueprint, stream_with_context, g
from quart import Quart, Response
from quart.views import MethodView
from quart.ctx import copy_current_request_context

blueprint = Blueprint('main', __name__)


class StreamTestResource(MethodView):
    async def get(self):
        g.username = 'test'

        # 复制当前请求上下文到生成器
        @copy_current_request_context
        @stream_with_context
        async def async_generator():
            print(g.username)
            for i in range(5):
                await asyncio.sleep(1)
                time = datetime.isoformat(datetime.now())
                yield f'data: {time}\n\n'

        return Response(
            async_generator(),
            200,
            headers={
                'Content-Type': 'text/event-stream',
                'Cache-Control': 'no-cache',
                'Transfer-Encoding': 'chunked',
            }
        )


blueprint.add_url_rule('/ping', view_func=StreamTestResource.as_view('ping'))

app = Quart(__name__)
app.register_blueprint(blueprint)

if __name__ == '__main__':
    app.run()

内容的提问来源于stack exchange，提问作者Yash Khandelwal