Published 2021-08-01 by Seth Larson
Reading time: 6 minutes
Starting in v2.0 Flask has added async views
which allow using async and await within a view function. This allows you to use other async
APIs when building a web application with Flask.
If you're planning on using Flask's async views there's a consideration to be aware of for using globally defined API clients or fixtures that are async.
In this example we're using the async Elasticsearch Python client as our global fixture. We initialize a simple Flask application with a single async view that makes a request with the async Elasticsearch client:
# app.py
from flask import Flask, jsonify
from elasticsearch import AsyncElasticsearch
app = Flask(__name__)
# Create the AsyncElasticsearch instance in the global scope.
es = AsyncElasticsearch(
"https://localhost:9200",
api_key="..."
)
@app.route("/", methods=["GET"])
async def async_view():
return jsonify(**(await es.info()))
Running with gunicorn via $gunicorn app:app and then visiting the app via http://localhost:8000.
After the first request everything looks fine:
{
"cluster_name": "d31d9d6abb334a398210484d7ac8567b",
"cluster_uuid": "K5uyniiMT9u2grNBmsSt_Q",
"name": "instance-0000000001",
"tagline": "You Know, for Search",
"version": {
"build_date": "2021-04-20T20:56:39.040728659Z",
"build_flavor": "default",
"build_hash": "3186837139b9c6b6d23c3200870651f10d3343b7",
"build_snapshot": false,
"build_type": "docker",
"lucene_version": "8.8.0",
"minimum_index_compatibility_version": "6.0.0-beta1",
"minimum_wire_compatibility_version": "6.8.0",
"number": "7.13.1"
}
}
However when you refresh the page to send a second request you receive
an InternalError and the following traceback:
Traceback (most recent call last):
...
File "/app/app.py", line 13, in async_route
return jsonify(**(await es.info()))
File "/app/venv/lib/...", line 288, in info
return await self.transport.perform_request(
File "/app/venv/lib/...", line 327, in perform_request
raise e
File "/app/venv/lib/...", line 296, in perform_request
status, headers, data = await connection.perform_request(
File "/app/venv/lib/...", line 312, in perform_request
raise ConnectionError("N/A", str(e), e)
ConnectionError(Event loop is closed) caused by:
RuntimeError(Event loop is closed)
The error message mentions the event loop is closed, huh? To understand why this is happening you
need to know how AsyncElasticsearch is implemented and how async views in Flask work.
Async code relies on something called an event loop. So any code using async or await can't execute
without an event loop that is "running". The unfortunate thing is that there's no running event
loop right when you start Python (ie, the global scope).
This is why you can't have code that looks like this:
async def f():
print("I'm async!")
# Can't do this!
await f()
instead you have to use asyncio.run() and typically an async main/entrypoint function to use await like so:
import asyncio
async def f():
print("I'm async!")
async def main():
await f()
# asyncio starts an event loop here:
asyncio.run(main())
(There's an exception to this via python -m asyncio / IPython, but really this is running the REPL after starting an event loop)
So if you need an event loop to run any async code, how can you define an
AsyncElasticsearch instance in the global scope?
The magic of global definitions for AsyncElasticsearch is delaying the full initialization
of calling asyncio.get_running_loop(), creating aiohttp.Session, sniffing, etc
until after we've received our first async call. Once an async call is made
we can almost guarantee that there's a running event loop, because if there
wasn't a running event loop the request wouldn't work out anyways.
This is great for async programs especially, as typically a single event loop gets
used throughout a single execution of the program and means you can create your AsyncElasticsearch
instance in the global scope how users create their synchronous Elasticsearch client in the global scope.
Using multiple event loops is tricky and would likely break many other libraries like aiohttp
in the process for no (?) benefit, so we don't support this configuration. Now
how does this break when used with Flask's new async views?
The simple explanation is that Flask uses WSGI to service HTTP requests and responses which doesn't support asynchronous I/O. Asynchronous code requires a running event loop to execute, so Flask needs to get a running event loop from somewhere in order to execute an async view.
To do so, Flask will create a new event loop and start running the view within this new event loop for every execution of the async view. This means all the async and await calls within the view will see the same event loop, but any other request before or after this view will see a different event loop.
The trouble comes when you want to use async fixtures that are in the global scope, which in my experience is common in small to medium Flask applications. Very unfortunate situation! So what can we do?
The problem isn't with Flask or the Python Elasticsearch client, the problem is the incompatibility between WSGI and async globals. There are a couple of solutions, both of which involve Async Server Gateway Interface (ASGI), WSGI's async-flavored cousin which was designed with async programs in mind.
One way to avoid the problem with WSGI completely is to simply use a native ASGI web application framework instead. There are a handful of popular and widely used ASGI frameworks you can choose from:
If you're looking for an experience that's very similar to Flask you can use Quart which is inspired by Flask. Quart even has a guide about how to migrate from a Flask application to using Quart! Flask's own documentation for async views actually recommends using Quart in some cases due to the performance hit from using a new event loop per request.
If you're looking to learn something new you can check out FastAPI which includes a bunch of builtin functionality for documenting APIs, strict model declarations, and data validation.
Something to keep in mind when developing an ASGI application is you need an ASGI-compatible server. Common choices include Uvicorn, Hypercorn, and Daphne. Another option is to use the Gunicorn with Uvicorn workers.
All the options mentioned above function pretty similarly so pick whichever one you like. My personal choice has historically been Gunicorn with Uvicorn workers because of how widely used and mature Gunicorn is relative to how new the other libraries are.
You can do so like this:
$ gunicorn app:app -k uvicorn.workers.UvicornWorker
If you really love Flask and want to continue using it you can also use
the asgiref package provides an easy wrapper
called WsgiToAsgi that converts a WSGI application to an ASGI application.
from flask import Flask, jsonify
from elasticsearch import AsyncElasticsearch
# Same definition as above...
wsgi_app = Flask(__name__)
es = AsyncElasticsearch(
"https://localhost:9200",
api_key="..."
)
@wsgi_app.route("/", methods=["GET"])
async def async_view():
return jsonify(**(await es.info()))
# Convert the WSGI application to ASGI
from asgiref.wsgi import WsgiToAsgi
asgi_app = WsgiToAsgi(wsgi_app)
In this example we're converting the WSGI application wsgi_app into an ASGI application asgi_app
which means when we run the application a single event loop will be used for every request
instead of a new event loop per request.
This approach will still require you to use an ASGI-compatible server.
Wow, you made it to the end!
If you're like me, you don't believe social media should be the way to get updates on the cool stuff your friends are up to. Instead, you should either follow my blog with the RSS reader of your choice or via my email newsletter for guaranteed article publication notifications.
If you really enjoyed a piece I would be grateful if you shared with a friend. If you have follow-up thoughts you can send them via email.
Thanks for reading!
— Seth