Async in Flask 2.0

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

Contents

Flask 2.0 Async

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:

1. You just need to install Flask with the extra async via pip install "Flask[async]".
2. Then, you can add the async keyword 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. In other words, while you can execute various async tasks in a single request, each async task must finish before a response gets sent back. 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 I/O-bound operations, when both of these are true:

1. There's a number of operations
2. Each operation takes less than a few seconds to finish

For example:

1. making HTTP or API calls
2. interacting with a database
3. working with the file system

It's not appropriate for background and long-running tasks as well as cpu-bound operations, like:

1. Running machine learning models
2. Processing images or PDFs
3. Performing backups

Such tasks would be better implemented using a task queue like Celery to manage separate long-running tasks.

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 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 = ['https://www.kennedyrecipes.com',
        'https://www.kennedyrecipes.com/breakfast/pancakes/',
        'https://www.kennedyrecipes.com/breakfast/honey_bran_muffins/']

# 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:

1. Create multiple asynchronous tasks (asyncio.create_task())
2. 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, 'https://www.kennedyrecipes.com/baked_goods/bagels/')

    assert str(result['url']) == 'https://www.kennedyrecipes.com/baked_goods/bagels/'
    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.