text: Flask 2.0, which was released on May 11th, 2021, adds built-in support for asynchronous routes, error handlers, before and after request functions, and teardown callbacks!
  This article looks at Flask 2.0's new async functionality and how to leverage it in your Flask projects.
  This article assumes that you have prior experience with Flask. If you're interested in learning more about Flask, check out my course on how to build, test, and deploy a Flask application:
  Developing Web Applications with Python and Flask
  Starting with Flask 2.0, you can create asynchronous route handlers using
  async/
  await:
  import asyncio async def async_get_data(): await asyncio.sleep(1) return 'Done!' @app.route("/data") async def get_data(): data = await async_get_data() return data
  Creating asynchronous routes is as simple as creating a synchronous route:
  - You just need to install Flask with the extra
  asyncvia
  pip install "Flask[async]".
  - Then, you can add the
  asynckeyword to your functions and use
  await.
  How Does this Work?
  The following diagram illustrates how asynchronous code is executed in Flask 2.0:
  In order to run asynchronous code in Python, an event loop is needed to run the coroutines. Flask 2.0 takes care of creating the asyncio event loop -- typically done with
  asyncio.run() -- for running the coroutines.
  If you're interested in learning more about the differences between threads, multiprocessing, and async in Python, check out the Speeding Up Python with Concurrency, Parallelism, and asyncio post.
  When an
  async route function is processed, a new sub-thread will be created. Within this sub-thread, an asyncio event loop will execute to run the route handler (coroutine).
  This implementation leverages the
  asgiref library (specifically the AsyncToSync functionality) used by Django to run asynchronous code.
  For more implementation specifics, refer to
  async_to_sync()in the Flask source code.
  What makes this implementation great is that it allows Flask to be run with any worker type (threads, gevent, eventlet, etc.).
  Running asynchronous code prior to Flask 2.0 required creating a new asyncio event loop within each route handler, which necessitated running the Flask app using thread-based workers. More details to come later in this article...
  Additionally, the use of asynchronous route handlers is backwards-compatible. You can use any combination of async and sync route handlers in a Flask app without any performance hit. This allows you to start prototyping a single async route handler right away in an existing Flask project.
  Why is ASGI not required?
  By design, Flask is a synchronous web framework that implements the WSGI (Web Server Gateway Interface) protocol.
  WSGI is an interface between a web server and a Python-based web application. A WSGI (Web Server Gateway Interface) server (such as Gunicorn or uWSGI) is necessary for Python web applications since a web server cannot communicate directly with Python.
  Want to learn more about WSGI?
  Check out 'What is Gunicorn in Python?' and take a look at the Building a Python Web Framework course.
  When processing requests in Flask, each request is handled individually within a worker. The asynchronous functionality added to Flask 2.0 is always within a single request being handled:
  Keep in mind that even though asynchronous code can be executed in Flask, it's executed within the context of a synchronous framework. Therefore, there are limited situations where asynchronous routes will actually be beneficial. There are other Python web frameworks that support ASGI (Asynchronous Server Gateway Interface), which supports asynchronous call stacks so that routes can run concurrently:
  When Should Async Be Used?
  While asynchronous execution tends to dominate discussions and generate headlines, it's not the best approach for every situation.
  It's ideal for Input/Output (I/O) processing (I/O-bound) when there's a lot of calls to external sources:
  - servers (HTTP or API calls)
  - databases
  - file systems
  Asynchronous HTTP calls
  The asynchronous approach really pays dividends when you need to make multiple HTTP requests to an external website or API. For each request, there will be a significant amount of time needed for the response to be received. This wait time translates to your web app feeling slow or sluggish to your users.
  Instead of making external requests one at a time (via the requests package), you can greatly speed up the process by leveraging
  async/
  await.
  In the synchronous approach, an external API call (such as a GET) is made and then the application waits to get the response back. The amount of time it takes to get a response back is called latency, which varies based on Internet connectivity and server response times. Latency in this case will probably be in the 0.2 - 1.5 second range per request.
  In the asynchronous approach, an external API call is made and then processing continues on to make the next API call. As soon as a response is received from the external server, it's processed. This is a much more efficient use of resources.
  In general, asynchronous programming is perfect for situations like this where multiple external calls are made and there's a lot of waiting for I/O responses.
  Async Route Handler
  aiohttp is a package that uses asyncio to to create asynchronous HTTP clients and servers. If you're familiar with the requests package for performing HTTP calls synchronously, aiohttp is a similar package that focuses on asynchronous HTTP calls.
  Here's an example of aiohttp being used in a Flask route:
  urls = ['', '', ''] # Helper Functions async def fetch_url(session, url): """Fetch the specified URL using the aiohttp session specified.""" response = await session.get(url) return {'url': response.url, 'status': response.status} # Routes @app.route('/async_get_urls_v2') async def async_get_urls_v2(): """Asynchronously retrieve the list of URLs.""" async with ClientSession() as session: tasks = [] for url in urls: task = asyncio.create_task(fetch_url(session, url)) tasks.append(task) sites = await asyncio.gather(*tasks) # Generate the HTML response response = '<h1>URLs:</h1>' for site in sites: response += f"<p>URL: {site['url']} --- Status Code: {site['status']}</p>" return response
  You can find the source code for this example in the flask-async repo on GitLab.
  The
  async_get_urls_v2() coroutine uses a common asyncio pattern:
  - Create multiple asynchronous tasks (
  asyncio.create_task())
  - Run them concurrently (
  asyncio.gather())
  Testing Async Routes
  You can test an async route handler just like you normally would with pytest since Flask handles all the async processing:
  @pytest.fixture(scope='module') def test_client(): # Create a test client using the Flask application with app.test_client() as testing_client: yield testing_client # this is where the testing happens! def test_async_get_urls_v2(test_client): """ GIVEN a Flask test client WHEN the '/async_get_urls_v2' page is requested (GET) THEN check that the response is valid """ response = test_client.get('/async_get_urls_v2') assert response.status_code == 200 assert b'URLs' in response.data
  This is a basic check for a valid response from the
  /async_get_urls_v2 URL using the
  test_client fixture.
  More Async Examples
  Request callbacks can also be async in Flask 2.0:
  # Helper Functions async def load_user_from_database(): """Mimics a long-running operation to load a user from an external database.""" app.logger.info('Loading user from database...') await asyncio.sleep(1) async def log_request_status(): """Mimics a long-running operation to log the request status.""" app.logger.info('Logging status of request...') await asyncio.sleep(1) # Request Callbacks @app.before_request async def app_before_request(): await load_user_from_database() @app.after_request async def app_after_request(response): await log_request_status() return response
  Error handlers as well:
  # Helper Functions async def send_error_email(error): """Mimics a long-running operation to log the error.""" app.logger.info('Logging status of error...') await asyncio.sleep(1) # Error Handlers @app.errorhandler(500) async def internal_error(error): await send_error_email(error) return '500 error', 500
  Flask 1.x Async
  You can mimic Flask 2.0 async support in Flask 1.x by using
  asyncio.run() to manage the asyncio event loop:
  # Helper Functions async def fetch_url(session, url): """Fetch the specified URL using the aiohttp session specified.""" response = await session.get(url) return {'url': response.url, 'status': response.status} async def get_all_urls(): """Retrieve the list of URLs asynchronously using aiohttp.""" async with ClientSession() as session: tasks = [] for url in urls: task = asyncio.create_task(fetch_url(session, url)) tasks.append(task) results = await asyncio.gather(*tasks) return results # Routes @app.route('/async_get_urls_v1') def async_get_urls_v1(): """Asynchronously retrieve the list of URLs (works in Flask 1.1.x when using threads).""" sites = asyncio.run(get_all_urls()) # Generate the HTML response response = '<h1>URLs:</h1>' for site in sites: response += f"<p>URL: {site['url']} --- Status Code: {site['status']}</p>" return response
  The
  get_all_urls() coroutine implements similar functionality that was covered in the
  async_get_urls_v2() route handler.
  How does this work?
  In order for the asyncio event loop to properly run in Flask 1.x, the Flask application must be run using threads (default worker type for Gunicorn, uWSGI, and the Flask development server):
  Each thread will run an instance of the Flask application when a request is processed. Within each thread, a separate asyncio event loop is created for running any asynchronous operations.
  Testing Coroutines
  You can use pytest-asyncio to test asynchronous code like so:
  @pytest.mark.asyncio async def test_fetch_url(): """ GIVEN an `asyncio` event loop WHEN the `fetch_url()` coroutine is called THEN check that the response is valid """ async with aiohttp.ClientSession() as session: result = await fetch_url(session, '') assert str(result['url']) == '' assert int(result['status']) == 200
  This test function uses the
  @pytest.mark.asyncio decorator, which tells pytest to execute the coroutine as an asyncio task using the asyncio event loop.
  Conclusion
  The asynchronous support added in Flask 2.0 is an amazing feature! However, asynchronous code should only be used when it provides an advantage over the equivalent synchronous code. As you saw, one example of when asynchronous execution makes sense is when you have to make multiple HTTP calls within a route handler.
  --
  I performed some timing tests using the Flask 2.0 asynchronous function (
  async_get_urls_v2()) vs. the equivalent synchronous function. I performed ten calls to each route:
  The asynchronous version is about 8x faster! So, if you have to make multiple external HTTP calls within a route handler, the increased complexity of using asyncio and aiohttp is definitely justified based on the significant decrease in execution time.
  If you'd like to learn more about Flask, be sure to check out my course -- Developing Web Applications with Python and Flask.