I am currently building an app with Quart that web scrapes an external site for data. Because this can take several minutes, I made the scraping a background task to avoid request timeouts.
After the user provides a user ID to a form on the home page, the app sets a background task to scrape the data and in the meantime, the user is redirected to a loading page.
@app.route('/')
async def index():
form = await request.form
id = int(form['user_id'])
# set the background task
app.add_background_task(task_to_do, user_id=id)
return redirect('/loading')
@app.route('/loading')
async def loading():
return await render_template('loading.html')
Currently, the user is successfully redirected to the loading page and the scraping background task finishes no problem.
At the completion of my task_to_do
background task, I want the user to be redirected to a new page. How would one go about doing that?
I think you want to utilise Server Sent Events. This is where the client creates a connection to the server and waits for events from it. You could first send a loading event, and then when the computation completes send a result event. There is a guide in the Quart docs for Server Sent Events.
E.g. something like the code below (using the ServerSentEvent class from the docs). Note the user_id needs to be a path parameter as SSE works over GET requests.
from quart import abort, make_response
@app.get('/<int:id_>')
async def index(id_):
if "text/event-stream" not in request.accept_mimetypes:
abort(400)
async def send_events():
event = ServerSentEvent("loading")
yield event.encode()
data = await task_to_do(user_id=id_)
event = ServerSentEvent(json.dumps(data))
yield event.encode()
response = await make_response(
send_events(),
{
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Transfer-Encoding': 'chunked',
},
)
response.timeout = None
return response