Flask can run asynchronous view functions, but it’s not immediately obvious how to make it happen.
Here’s a simple Flask app with an async view function:
import asyncio
from flask import Flask
app = Flask(__name__)
async def fetch_data_from_external_api():
"""Simulates fetching data from an external API."""
await asyncio.sleep(2) # Simulate network latency
return {"data": "some fetched data"}
@app.route("/api/data")
async def get_data():
"""Asynchronous view function to fetch and return data."""
data = await fetch_data_from_external_api()
return data
if __name__ == "__main__":
# For development, use the built-in server with an ASGI wrapper
# For production, use a proper ASGI server like uvicorn or hypercorn
from flask.cli import run_command
app.run(debug=True)
When you run this and visit /api/data, you’ll notice a 2-second delay before the JSON response appears. This delay is the asyncio.sleep(2) within fetch_data_from_external_api, demonstrating that the asynchronous operation is indeed being executed.
The core problem Flask solves here is how to integrate Python’s async/await syntax into a traditionally synchronous web framework. Before Flask 2.0, attempting to use async def for a view function would result in a TypeError: object ... can't be used in 'await' expression. Flask 2.0 and later, however, introduced native support for asynchronous view functions, but this support is mediated by an ASGI (Asynchronous Server Gateway Interface) server.
When you run app.run(debug=True), Flask uses its built-in development server. For async views to work, this development server needs to be run under an ASGI server environment. Flask’s app.run() in recent versions automatically handles this by leveraging a compatible ASGI server (like werkzeug’s built-in ASGI support or by defaulting to uvicorn if installed).
Here’s how it breaks down internally:
- ASGI Application: Flask applications can be wrapped to conform to the ASGI specification. This means Flask can be run by any ASGI server (like
uvicorn,hypercorn,daphne). async defViews: When Flask encounters anasync defview function, it doesn’t execute it directly. Instead, it treats the view function as an awaitable coroutine.- ASGI Server Orchestration: The ASGI server (or Flask’s development server acting as one) is responsible for managing the event loop. When an incoming request hits an
async defview, the ASGI server schedules this coroutine to run on the event loop. awaitOperations: Inside theasync defview, anyawaitcall (likeawait asyncio.sleep(2)orawait httpx.get(...)) yields control back to the event loop. This allows the server to handle other incoming requests or background tasks without blocking the entire application. Once the awaited operation completes, the event loop resumes the coroutine.- Response Generation: After the
async defview function completes (i.e., returns a response), the ASGI server packages this response and sends it back to the client.
The primary levers you control are the asynchronous operations within your view functions. You can await any awaitable object:
- I/O-bound tasks: Network requests (using
httpxoraiohttp), database queries (usingasyncpgoraiosqlite), file operations (usingaiofiles). asyncioprimitives:asyncio.sleep(),asyncio.gather(),asyncio.create_task().
Consider this example for concurrent fetching:
import asyncio
import httpx # You'll need to install this: pip install httpx
from flask import Flask, jsonify
app = Flask(__name__)
async def fetch_url(url):
async with httpx.AsyncClient() as client:
response = await client.get(url)
response.raise_for_status() # Raise an exception for bad status codes
return response.json()
@app.route("/api/concurrent_data")
async def get_concurrent_data():
urls = [
"https://jsonplaceholder.typicode.com/todos/1",
"https://jsonplaceholder.typicode.com/posts/1",
]
# Create tasks for each fetch operation
tasks = [fetch_url(url) for url in urls]
# Run tasks concurrently and wait for all to complete
results = await asyncio.gather(*tasks)
return jsonify(results)
if __name__ == "__main__":
app.run(debug=True)
When you hit /api/concurrent_data, the two requests to jsonplaceholder.typicode.com are made concurrently, not sequentially. asyncio.gather waits for both to finish. This is where async shines: reducing latency by overlapping I/O operations.
The one thing most people don’t realize is that app.run() in Flask 2.0+ doesn’t directly run async views. It uses its development server in a way that’s compatible with ASGI. When you deploy to production, you must use a dedicated ASGI server like uvicorn. You’d typically run it with a command like uvicorn main:app --reload (assuming your Flask app instance is named app in a file called main.py). The ASGI server is the true manager of the event loop and the async/await execution.
The next hurdle you’ll face is managing background tasks or long-running operations that shouldn’t block the request-response cycle, even in an async view.