I tried to build a webhook using the Python Flask framework. The app_producer.py file (client) sends HTTP POST requests, and the app_consumer.py file (server) processes requests sent by the app_producer.py file.
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
# 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', stockInfo = 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)
The consumer.html Jinja template file that I wanted to render:
<!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>
<script>
document.getElementById("consumeTasks").onclick = function(){
location.href = "http://localhost:5001/consumetasks";
};
</script>
</div>
<div id="stockSheet"> </div>
<script>
{% for stock in stockInfo %}
{% if loop.index0 is even %}
var SKU = "{{stock}}";
document.getElementById("stockSheet").innerHTML += ("Product " + SKU);
{% elif loop.index0 is odd %}
var stockStatus = "{{stock}}";
document.getElementById("stockSheet").innerHTML += (" is " + stockStatus + "<br>");
{% endif %}
{% endfor %}
</script>
</body>
</html>
In the consumer.html file, I used a <button> element to redirect the page from localhost:5001 to localhost:5001/consumetasks, so that the code in the Jinja template can be rendered. However, when I ran both the app_producer.py and app_consumer.py files to render the consumer.html template file and clicked the button on the localhost:5001 page, the contents inside the Jinja for loop failed to render.
When I checked the source code for the page localhost:5001/consumetasks, I found its source code was exactly the same as the code returned by the elif branch in the get_stock_status() function in app_consumer.py. I thought the problem was that the if branch in the get_stock_status() function in app_consumer.py was never executed. How can I fix my code so that the Jinja for loop in the consumer.html template can be rendered correctly?
)