英文:
“RuntimeError: working outside of request context" when using a generator to stream data with Flask
问题
我在尝试使用Python Flask框架构建Webhook时遇到了RuntimeError: working outside of request context错误。
这是发送HTTP POST请求的app_producer.py文件：
from flask import Response, render_template
from init_producer import app
import tasks_producer
# 以流的形式呈现具有给定上下文的模板并返回TemplateStream
def render_template_stream(template_name, **context):
app.update_template_context(context) # 使用一些常用变量更新模板上下文。
t = app.jinja_env.get_template(template_name) # 通过名称加载模板并返回Template。
rv = t.stream(context) # 返回一个TemplateStream，该流会逐个将函数作为字符串返回。
rv.enable_buffering(5) # 在产生之前缓冲5个项目
return rv # 返回TemplateStream
@app.route("/", methods=['GET'])
def index():
return render_template('producer.html')
@app.route('/producetasks', methods=['POST'])
def producetasks():
print("Producing tasks")
return Response(render_template_stream('producer.html', data = tasks_producer.produce_bunch_tasks()))
# 防止在将app_producer.py文件作为模块导入到另一个文件时自动执行app.run()函数。
if __name__ == "__main__":
app.run(host="localhost", port=5000, debug=True)
这是处理app_producer.py文件发送的请求的app_consumer.py文件：
from flask import render_template, Response, request
from flask_socketio import join_room
from init_consumer import app, socketio
import tasks_consumer
import uuid
# 以流的形式呈现具有给定上下文的模板并返回TemplateStream
def render_template_stream(template_name, **context):
app.update_template_context(context)
t = app.jinja_env.get_template(template_name)
rv = t.stream(context)
rv.enable_buffering(5)
return rv
# 在这个应用程序实例的第一个请求之前注册要运行的函数
# 创建唯一的会话ID并将其存储在应用程序配置文件中
@app.before_request
def initialize_params():
if not hasattr(app.config,'uid'):
sid = str(uuid.uuid4())
app.config['uid'] = sid
print("initialize_params - Session ID stored =", sid)
# 呈现分配的模板文件
@app.route("/", methods=['GET'])
def index():
return render_template('consumer.html', stockSheet = {})
@app.route('/consumetasks', methods=['GET','POST'])
def get_stock_status():
# 处理POST请求
if request.method == 'POST':
print("Retrieving stock status")
return Response(render_template_stream('consumer.html', stockSheet = tasks_consumer.sendStockStatus()))
# 处理GET请求
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>
'''
# 使用端口5001运行
if __name__ == "__main__":
socketio.run(app, host='localhost', port=5001, debug=True)
app_consumer.py文件导入的tasks_consumer.py文件：
import csv
from flask import request
from init_consumer import app, socketio
import json
# 接收Webhook请求并向客户端发出SocketIO事件
def send_message(data):
status_code = 0
if request.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 # 方法不允许
return status_code
# 检索通过Webhook请求发送的产品的库存状态并将其返回给客户端。
@app.route('/consumetasks', methods=['POST'])
def sendStockStatus():
request_data = request.get_json()
stockList = [] # 库存中的产品列表
stockInfo = [] # 请求中发送的产品列表
stockSheet = {} # 请求中发送的产品及其库存状态的字典
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)
print(stockSheet)
当我运行app_producer和app_consumer文件以呈现consumer.html模板文件时，tasks_consumer.py文件中的sendStockStatus()函数中的request_data = request.get_json()行引发了错误：RuntimeError: Working outside of request context。
让我看一下如何修复此错误。
英文:
I encountered the RuntimeError: working outside of request context error while trying to build a webhook using the Python Flask framework.
This is the app_producer.py file that sends HTTP POST requests:
from flask import Response, render_template
from init_producer import app
import tasks_producer
# Render a template with a given context as a stream and return a TemplateStream
def render_template_stream(template_name, **context):
app.update_template_context(context) # Update the template context with some commonly used variables.
t = app.jinja_env.get_template(template_name) # jinja2.Environment.get_template() # Load a template by name with loader and return a Template.
rv = t.stream(context) # jinja2.Template.stream # Return a TemplateStream that returns one function after another as strings
rv.enable_buffering(5) # jinja2.environment.TemplateStream.enable_buffering # Buffer 5 items before yielding them
return rv # Return a TemplateStream
@app.route("/", methods=['GET'])
def index():
return render_template('producer.html')
@app.route('/producetasks', methods=['POST'])
def producetasks():
print("Producing tasks")
return Response(render_template_stream('producer.html', data = tasks_producer.produce_bunch_tasks()))
# Stop the app.run() function from being automatically executed when the app_producer.py file is imported as a module to another file.
if __name__ == "__main__":
app.run(host="localhost",port=5000, debug=True)
This is the app_consumer.py file that processes requests sent by the app_producer.py file.
from flask import render_template, Response, request
from flask_socketio import join_room
from init_consumer import app, socketio
import tasks_consumer
import uuid
# Render a template with a given context as a stream and return a TemplateStream
def render_template_stream(template_name, **context):
app.update_template_context(context)
t = app.jinja_env.get_template(template_name)
rv = t.stream(context)
rv.enable_buffering(5)
return rv
# Registers a function to be run before the first request to this instance of the application
# Create a unique session ID and store it within the application configuration file
@app.before_request
def initialize_params():
if not hasattr(app.config,'uid'):
sid = str(uuid.uuid4())
app.config['uid'] = sid
print("initialize_params - Session ID stored =", sid)
# Render the assigned template file
@app.route("/", methods=['GET'])
def index():
return render_template('consumer.html', stockSheet = {})
@app.route('/consumetasks', methods=['GET','POST'])
def get_stock_status():
# Handle the POST request
if request.method == 'POST':
print("Retrieving stock status")
return Response(render_template_stream('consumer.html', stockSheet = tasks_consumer.sendStockStatus()))
# 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>
'''
# Run using port 5001
if __name__ == "__main__":
socketio.run(app,host='localhost', port=5001,debug=True)
The tasks_consumer.py file that the app_consumer.py file imports:
import csv
from flask import request
from init_consumer import app, socketio
import json
# Receive the webhook requests and emit a SocketIO event back to the client
def send_message(data):
status_code = 0
if request.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
# Retrieve the stock status of the products sent through the webhook requests and return them back to the client.
@app.route('/consumetasks', methods=['POST'])
def sendStockStatus():
request_data = request.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)
print(stockSheet)
When I ran both the app_producer and app_consumer files to render the consumer.html template file, the line request_data = request.get_json() in the sendStockStatus() function in the tasks_consumer.py file raised the error: RuntimeError: Working outside of request context.
What puzzles me is that I have put the sendStockStatus() function in a view function, but the body of the function still gave me the error Working outside of request context. Could anyone point me in the right direction as to how to fix the error?
答案1
得分: 2
当您使用生成器来实现Flask的流响应时，默认情况下不可用请求上下文。为了确保上下文在生成器运行时保持存在，您必须使用 stream_with_context() 包装生成器函数。
英文:
When you use a generator to implement a streaming response with Flask, the request context is not available by default. To ensure the context stays around while the generator runs, you have to wrap your generator function with the stream_with_context().
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论