Flask使用生成器流式传输数据时出现RuntimeError: working outside of request context

你的问题根源有两个：

1. 重复定义路由：app_consumer.py和tasks_consumer.py里都定义了/consumetasks的POST路由，这会导致Flask路由规则冲突，实际触发的是app_consumer里的路由，tasks_consumer里的路由根本没被使用。
2. 请求上下文未传递：你在app_consumer的get_stock_status里直接调用tasks_consumer.sendStockStatus()，此时sendStockStatus是作为普通生成器函数被调用，而非通过路由处理请求。Flask的request对象绑定到当前请求上下文，这种直接调用的方式无法获取到请求上下文，因此触发错误。

修复步骤 #

1. 移除tasks_consumer.py中的路由装饰器 #

sendStockStatus是被get_stock_status调用的工具函数，不需要作为独立路由存在，删掉它上面的@app.route('/consumetasks', methods=['POST'])装饰器。

2. 传递request对象到函数内部 #

Flask的request是上下文局部对象，不能在非路由处理函数里直接全局引用，需要把request作为参数传给sendStockStatus和send_message。

修改后的tasks_consumer.py代码：

import csv
from flask import request
from init_consumer import app, socketio
import json

# 修改：接收request参数
def send_message(data, req):
    status_code = 0
    if req.method == 'POST':
        roomid = app.config['uid']
        msg = json.dumps(data)
        event = "Send_stock_status"
        socketio.emit(event, msg, namespace = '/collectHooks', room = roomid)
        status_code = 200
    else:
        status_code = 405 # Method not allowed
    return status_code

# 修改：移除路由装饰器，接收request参数
def sendStockStatus(req):
    request_data = req.get_json()
    stockList = [] # List of products in stock
    stockInfo = [] # List of products sent in the request
    stockSheet = {} # Dictionary of products sent in the request and their stock status
    with open("NZ_NVJ_Apparel_SKUs_sheet.csv", newline='') as csvFile:
        stockReader = csv.reader(csvFile, delimiter=',', quotechar='"')
        for row in stockReader:
            stockList.append(row[0])
    
    if request_data:
        if 'SKU' in request_data:
            stockInfo = request_data['SKU']
            for stock in stockInfo:
                if stock in stockList:
                    stockStatus = "In Stock"
                    stockSheet.update({str(stock):stockStatus})
                    yield stock, stockStatus
                else:
                    stockStatus = "Out of Stock"
                    stockSheet.update({str(stock):stockStatus})
                    yield stock, stockStatus
    send_message(stockSheet, req)
    print(stockSheet)

3. 修改app_consumer.py中的调用方式 #

在get_stock_status的POST分支里，把当前的request对象传给sendStockStatus：

@app.route('/consumetasks', methods=['GET','POST'])
def get_stock_status():
    # Handle the POST request
    if request.method == 'POST':
        print("Retrieving stock status")
        # 修改：传递request参数
        return Response(render_template_stream('consumer.html', stockSheet = tasks_consumer.sendStockStatus(request)))     
    # Handle the GET request
    elif request.method == 'GET':
        return '''
         <!doctype html>
        <html>
            <head>
                <title>Stock Sheet</title>
                <meta name="viewport" content="width=device-width, initial-scale=1.0"/>
            </head>

            <body class="container">
                <h1>Stock Sheet</h1>
                <div>
                    <button id="consumeTasks">Check stock status</button>
                </div>
            </body>
        </html>
        '''

额外说明 #

• Flask的上下文对象（request、session等）只能在当前请求的路由处理函数内部直接使用，若要在其他函数里使用，要么传递上下文对象，要么手动创建上下文（适合测试，不推荐生产环境）。
• 避免重复定义路由，同一个URL的路由规则只能存在一个，否则Flask会以最后注册的为准，导致之前的路由失效。

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