Skip to main content

​
Overview

Flask supports async and await syntax, allowing you to write asynchronous view functions and handlers. This enables concurrent I/O operations while maintaining Flask’s simple programming model.
Flask’s async support is designed for WSGI servers. For true async performance, consider using an ASGI server with an async framework like Quart (Flask’s async cousin).

​
Basic Async Views

​
Simple Async View

from flask import Flask
import asyncio

app = Flask(__name__)

@app.route('/')
async def index():
    """Async view function."""
    await asyncio.sleep(1)
    return 'Hello, Async World!'

​
Async with Database

import aiohttp

@app.route('/users/<int:user_id>')
async def get_user(user_id):
    """Fetch user from async database."""
    user = await User.get(user_id)
    return jsonify(user.to_dict())

@app.route('/users', methods=['POST'])
async def create_user():
    """Create user asynchronously."""
    data = await request.get_json()
    user = await User.create(**data)
    return jsonify(user.to_dict()), 201

​
How It Works

​
The ensure_sync Method

Flask automatically wraps async functions to work with WSGI: 
from inspect import iscoroutinefunction

def ensure_sync(self, func):
    """Ensure that the function is synchronous for WSGI workers.
    Plain def functions are returned as-is. Async def functions 
    are wrapped to run and wait for the response.
    """
    if iscoroutinefunction(func):
        return self.async_to_sync(func)
    return func

​
The async_to_sync Method

def async_to_sync(self, func):
    """Return a sync function that will run the coroutine function.
    
    Requires the 'async' extra: pip install 'flask[async]'
    """
    from asgiref.sync import async_to_sync as asgiref_async_to_sync
    return asgiref_async_to_sync(func)

​
Installation

Install Flask with async support: 
pip install 'flask[async]'
This installs asgiref, which provides the async-to-sync adapter.

​
Async Patterns

​
Concurrent API Calls

import aiohttp
import asyncio

@app.route('/aggregate')
async def aggregate_data():
    """Fetch data from multiple APIs concurrently."""
    async with aiohttp.ClientSession() as session:
        # Run requests concurrently
        tasks = [
            fetch_users(session),
            fetch_posts(session),
            fetch_comments(session)
        ]
        users, posts, comments = await asyncio.gather(*tasks)
    
    return jsonify({
        'users': users,
        'posts': posts,
        'comments': comments
    })

async def fetch_users(session):
    async with session.get('https://api.example.com/users') as resp:
        return await resp.json()

async def fetch_posts(session):
    async with session.get('https://api.example.com/posts') as resp:
        return await resp.json()

async def fetch_comments(session):
    async with session.get('https://api.example.com/comments') as resp:
        return await resp.json()

​
Async Database Queries

from databases import Database

database = Database('postgresql://user:pass@localhost/db')

@app.before_serving
async def connect_db():
    await database.connect()

@app.after_serving
async def disconnect_db():
    await database.disconnect()

@app.route('/users')
async def list_users():
    """Query database asynchronously."""
    query = "SELECT * FROM users ORDER BY created_at DESC LIMIT 10"
    results = await database.fetch_all(query)
    return jsonify([dict(row) for row in results])

@app.route('/users/<int:user_id>')
async def get_user(user_id):
    query = "SELECT * FROM users WHERE id = :user_id"
    result = await database.fetch_one(query, {'user_id': user_id})
    if result is None:
        abort(404)
    return jsonify(dict(result))

​
Background Tasks

import asyncio
from datetime import datetime

@app.route('/send-email', methods=['POST'])
async def send_email():
    """Send email asynchronously."""
    data = await request.get_json()
    
    # Start background task
    asyncio.create_task(send_email_async(
        to=data['to'],
        subject=data['subject'],
        body=data['body']
    ))
    
    return jsonify({'status': 'Email queued'}), 202

async def send_email_async(to, subject, body):
    """Actually send the email."""
    await asyncio.sleep(2)  # Simulate email sending
    print(f'Email sent to {to}')

​
Streaming Responses

import asyncio

@app.route('/stream')
async def stream():
    """Stream data asynchronously."""
    async def generate():
        for i in range(10):
            await asyncio.sleep(1)
            yield f'data: {i}\n\n'
    
    return generate(), {'Content-Type': 'text/event-stream'}

​
Async Class-Based Views

​
MethodView with Async

from flask.views import MethodView

class UserAPI(MethodView):
    async def get(self, user_id):
        """Async GET handler."""
        if user_id is None:
            users = await User.all()
            return jsonify([u.to_dict() for u in users])
        
        user = await User.get(user_id)
        if user is None:
            abort(404)
        return jsonify(user.to_dict())
    
    async def post(self):
        """Async POST handler."""
        data = await request.get_json()
        user = await User.create(**data)
        return jsonify(user.to_dict()), 201
    
    async def put(self, user_id):
        """Async PUT handler."""
        user = await User.get(user_id)
        if user is None:
            abort(404)
        
        data = await request.get_json()
        await user.update(**data)
        return jsonify(user.to_dict())
    
    async def delete(self, user_id):
        """Async DELETE handler."""
        user = await User.get(user_id)
        if user is None:
            abort(404)
        
        await user.delete()
        return '', 204

# Register the view
user_view = UserAPI.as_view('user_api')
app.add_url_rule('/users', defaults={'user_id': None},
                 view_func=user_view, methods=['GET'])
app.add_url_rule('/users', view_func=user_view, methods=['POST'])
app.add_url_rule('/users/<int:user_id>', view_func=user_view,
                 methods=['GET', 'PUT', 'DELETE'])

​
Request Hooks

Async request hooks are supported: 
@app.before_request
async def before_request():
    """Async before request hook."""
    g.start_time = time.time()
    g.db = await database.connect()

@app.after_request
async def after_request(response):
    """Async after request hook."""
    duration = time.time() - g.start_time
    response.headers['X-Request-Duration'] = str(duration)
    await g.db.close()
    return response

@app.teardown_request
async def teardown_request(exception=None):
    """Async teardown hook."""
    if hasattr(g, 'db'):
        await g.db.close()

​
Error Handlers

Async error handlers work too: 
@app.errorhandler(404)
async def not_found(error):
    """Async error handler."""
    await log_error(error)
    return jsonify({'error': 'Not found'}), 404

@app.errorhandler(Exception)
async def handle_exception(error):
    """Async exception handler."""
    await send_error_to_monitoring(error)
    return jsonify({'error': 'Internal server error'}), 500

​
Limitations

​
WSGI vs ASGI

Flask runs on WSGI, which is synchronous: 
# Flask converts async to sync automatically
@app.route('/')
async def index():
    # This runs in a thread pool, not true async
    await asyncio.sleep(1)
    return 'Hello'
For true async performance, use ASGI with Quart: 
# Quart (Flask's async cousin) on ASGI
from quart import Quart

app = Quart(__name__)

@app.route('/')
async def index():
    # True async on ASGI
    await asyncio.sleep(1)
    return 'Hello'

​
Background Tasks

Background tasks created with asyncio.create_task may not complete: 
# Bad: Task may be cancelled when request ends
@app.route('/process')
async def process():
    asyncio.create_task(long_running_task())
    return 'Started'

# Good: Use a task queue instead
@app.route('/process')
def process():
    celery.send_task('long_running_task')
    return 'Started'

​
Best Practices

​
1. Use Async for I/O Operations

# Good: Async for I/O-bound operations
@app.route('/data')
async def get_data():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/data') as resp:
            return await resp.json()

# Bad: Async for CPU-bound operations
@app.route('/compute')
async def compute():
    # This blocks other requests!
    result = expensive_computation()
    return str(result)

​
2. Mix Sync and Async Carefully

# You can have both sync and async views
@app.route('/sync')
def sync_view():
    return 'Sync response'

@app.route('/async')
async def async_view():
    await asyncio.sleep(1)
    return 'Async response'

​
3. Handle Exceptions Properly

@app.route('/safe')
async def safe_operation():
    try:
        result = await risky_operation()
        return jsonify(result)
    except Exception as e:
        app.logger.error(f'Error: {e}')
        abort(500)

​
4. Use Timeouts

import asyncio

@app.route('/with-timeout')
async def with_timeout():
    try:
        result = await asyncio.wait_for(
            slow_operation(),
            timeout=5.0
        )
        return jsonify(result)
    except asyncio.TimeoutError:
        abort(504)  # Gateway Timeout

​
5. Consider Using Quart for True Async

For applications that benefit from async, consider Quart: 
pip install quart

from quart import Quart, jsonify

app = Quart(__name__)

@app.route('/')
async def index():
    # True async with ASGI
    return jsonify({'message': 'Hello'})

if __name__ == '__main__':
    app.run()

​
Testing Async Views

import pytest

@pytest.mark.asyncio
async def test_async_view(client):
    """Test async view."""
    response = await client.get('/async')
    assert response.status_code == 200

def test_async_view_sync(client):
    """Test async view with sync client."""
    # Flask test client works with async views
    response = client.get('/async')
    assert response.status_code == 200

​
When to Use Async

​
Good Use Cases

  • Making multiple external API calls
  • Database queries (with async DB driver)
  • File I/O operations
  • Network operations
  • WebSocket connections
  • CPU-intensive calculations
  • Simple CRUD operations
  • Applications with mostly synchronous code
  • When you need maximum compatibility

Build docs developers (and LLMs) love

Get started for freeTalk to us