fastapi-async-patterns

name	fastapi-async-patterns
user-invocable	false
description	Use when FastAPI async patterns for building high-performance APIs. Use when handling concurrent requests and async operations.
allowed-tools	["Bash","Read"]

FastAPI Async Patterns

Master async patterns in FastAPI for building high-performance, concurrent APIs with optimal resource usage.

Basic Async Route Handlers

Understanding async vs sync endpoints in FastAPI.

from fastapi import FastAPI
import time
import asyncio

app = FastAPI()

# Sync endpoint (blocks the event loop)
@app.get('/sync')
def sync_endpoint():
    time.sleep(1)  # Blocks the entire server
    return {'message': 'Completed after 1 second'}

# Async endpoint (non-blocking)
@app.get('/async')
async def async_endpoint():
    await asyncio.sleep(1)  # Other requests can be handled
    return {'message': 'Completed after 1 second'}

# CPU-bound work (use sync)
@app.get('/cpu-intensive')
def cpu_intensive():
    result = sum(i * i for i in range(10000000))
    return {'result': result}

# I/O-bound work (use async)
@app.get('/io-intensive')
async def io_intensive():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com/data')
        return response.json()

Async Database Operations

Async database patterns with popular ORMs and libraries.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select
import asyncpg
from motor.motor_asyncio import AsyncIOMotorClient
from tortoise import Tortoise
from tortoise.contrib.fastapi import register_tortoise

app = FastAPI()

# SQLAlchemy async setup
DATABASE_URL = 'postgresql+asyncpg://user:pass@localhost/db'
engine = create_async_engine(DATABASE_URL, echo=True, future=True)
AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

@app.get('/users/{user_id}')
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if  user:
         HTTPException(status_code=, detail=)
     user


  ():
    pool =  asyncpg.create_pool(
        ,
        min_size=,
        max_size=
    )
    :
         pool
    :
         pool.close()


  ():
      pool.acquire()  conn:
        row =  conn.fetchrow(
            , user_id
        )
          row:
             HTTPException(status_code=, detail=)
         (row)


mongo_client = AsyncIOMotorClient()
db = mongo_client.mydatabase


  ():
    document =  db.collection.find_one({: doc_id})
      document:
         HTTPException(status_code=, detail=)
     document


  ():
    result =  db.collection.insert_one(data)
     {: (result.inserted_id)}


register_tortoise(
    app,
    db_url=,
    modules={: []},
    generate_schemas=,
    add_exception_handlers=,
)

 tortoise.models  Model
 tortoise  fields

 ():
     = fields.IntField(pk=)
    name = fields.CharField(max_length=)
    email = fields.CharField(max_length=)


  ():
    user =  UserModel.get_or_none(=user_id)
      user:
         HTTPException(status_code=, detail=)
     user

Background Tasks

Fire-and-forget tasks without blocking the response.

from fastapi import BackgroundTasks, FastAPI
import asyncio
from datetime import datetime

app = FastAPI()

# Simple background task
async def send_email(email: str, message: str):
    await asyncio.sleep(2)  # Simulate email sending
    print(f'Email sent to {email}: {message}')

@app.post('/send-email')
async def send_email_endpoint(
    email: str,
    message: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(send_email, email, message)
    return {'status': 'Email will be sent in background'}

# Multiple background tasks
async def log_activity(user_id: int, action: str):
    await asyncio.sleep(0.5)
    print(f'[{datetime.now()}] User {user_id} performed: {action}')

async def update_analytics(action: str):
    await asyncio.sleep()
    ()


  ():
    
    background_tasks.add_task(log_activity, user_id, action)
    background_tasks.add_task(update_analytics, action)
     {: }


  ():
     asyncio.sleep()  
     os
     os.path.exists(file_path):
        os.remove(file_path)
        ()


  ():
    temp_path = 
     (temp_path, )  f:
        content =  file.read()
        f.write(content)

    
    background_tasks.add_task(cleanup_temp_files, temp_path)
     {: file.filename, : temp_path}

WebSocket Handling

Real-time bidirectional communication patterns.

from fastapi import WebSocket, WebSocketDisconnect, Depends
from typing import List
import json

app = FastAPI()

# Simple WebSocket
@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f'Echo: {data}')
    except WebSocketDisconnect:
        print('Client disconnected')

# WebSocket with authentication
async def get_current_user_ws(websocket: WebSocket):
    token = websocket.query_params.get('token')
    if not token or not verify_token(token):
        await websocket.close(code=1008)  # Policy violation
        raise HTTPException(status_code=401, detail='Unauthorized')
    return decode_token(token)

@app.websocket('/ws/authenticated')
async def authenticated_websocket(
    websocket: WebSocket,
    user = Depends()
):
     websocket.accept()
    :
         websocket.send_text()
         :
            data =  websocket.receive_text()
             websocket.send_text()
     WebSocketDisconnect:
        ()


 :
     ():
        .active_connections: [WebSocket] = []

      ():
         websocket.accept()
        .active_connections.append(websocket)

     ():
        .active_connections.remove(websocket)

      ():
         websocket.send_text(message)

      ():
         connection  .active_connections:
             connection.send_text(message)

manager = ConnectionManager()


  ():
     manager.connect(websocket)
     manager.broadcast()
    :
         :
            data =  websocket.receive_text()
             manager.broadcast()
     WebSocketDisconnect:
        manager.disconnect(websocket)
         manager.broadcast()



  ():
     websocket.accept()
    :
         :
            data =  websocket.receive_json()
            message_type = data.get()

             message_type == :
                 websocket.send_json({: })
             message_type == :
                 websocket.send_json({
                    : ,
                    : 
                })
     WebSocketDisconnect:
        ()

Server-Sent Events (SSE)

One-way streaming from server to client.

from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
import asyncio

app = FastAPI()

@app.get('/sse')
async def sse_endpoint():
    async def event_generator():
        for i in range(10):
            await asyncio.sleep(1)
            yield {
                'event': 'message',
                'data': f'Message {i}'
            }

    return EventSourceResponse(event_generator())

# SSE with real-time updates
@app.get('/sse/updates')
async def sse_updates():
    async def update_generator():
        while True:
            # Simulate fetching updates
            await asyncio.sleep(2)
            update = await fetch_latest_update()
            yield {
                'event': 'update',
                'data': json.dumps(update)
            }

    return EventSourceResponse(update_generator())

# SSE with heartbeat

  ():
      ():
        :
             :
                 asyncio.sleep()
                 {
                    : ,
                    : datetime.now().isoformat()
                }
         asyncio.CancelledError:
            ()

     EventSourceResponse(heartbeat_generator())

Streaming Responses

Stream large files or generated content.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import io
import csv

app = FastAPI()

# Stream large file
@app.get('/download/{filename}')
async def download_file(filename: str):
    async def file_stream():
        with open(f'/data/{filename}', 'rb') as f:
            while chunk := f.read(8192):
                yield chunk

    return StreamingResponse(
        file_stream(),
        media_type='application/octet-stream',
        headers={'Content-Disposition': f'attachment; filename={filename}'}
    )

# Stream generated CSV
@app.get('/export/users')
async def export_users():
    async def csv_stream():
        output = io.StringIO()
        writer = csv.writer(output)

        # Write header
        writer.writerow(['ID', 'Name', 'Email'])
        yield output.getvalue()
        output.truncate(0)
        output.seek()

        
        offset = 
        batch_size = 
         :
            users =  fetch_users_batch(offset, batch_size)
              users:
                

             user  users:
                writer.writerow([user., user.name, user.email])
                 output.getvalue()
                output.truncate()
                output.seek()

            offset += batch_size

     StreamingResponse(
        csv_stream(),
        media_type=,
        headers={: }
    )



  ():
      ():
         

         section  [, , ]:
             asyncio.sleep()  
            data =  fetch_section_data(section)
             .encode()
             .encode()

         

     StreamingResponse(report_stream(), media_type=)

Concurrent Request Handling

Parallel processing patterns for multiple operations.

from fastapi import FastAPI
import asyncio
import httpx

app = FastAPI()

# Parallel API calls
@app.get('/aggregate/user/{user_id}')
async def aggregate_user_data(user_id: int):
    async with httpx.AsyncClient() as client:
        # Fetch from multiple sources in parallel
        profile_task = client.get(f'https://api.example.com/users/{user_id}')
        posts_task = client.get(f'https://api.example.com/users/{user_id}/posts')
        comments_task = client.get(f'https://api.example.com/users/{user_id}/comments')

        profile, posts, comments = await asyncio.gather(
            profile_task,
            posts_task,
            comments_task
        )

        return {
            'profile': profile.json(),
            'posts': posts.json(),
            'comments': comments.json()
        }

# Parallel database queries
@app.get('/dashboard')
async def get_dashboard(db: AsyncSession = Depends(get_db)):
    # Execute multiple queries in parallel
    users_query = db.execute(select(User).limit(10))
    orders_query = db.execute(select(Order).limit(10))
    stats_query = db.execute(select(func.count(User.id)))

    users, orders, stats = await asyncio.gather(
        users_query,
        orders_query,
        stats_query
    )

     {
        : users.scalars().(),
        : orders.scalars().(),
        : stats.scalar()
    }



  ():
      httpx.AsyncClient()  client:
        tasks = [
            client.get(),
            client.get(),
            client.get()
        ]

        done, pending =  asyncio.wait(
            tasks,
            return_when=asyncio.FIRST_COMPLETED
        )

        
         task  pending:
            task.cancel()

        result = done.pop().result()
         result.json()

Async Context Managers

Resource management with async context managers.

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio

# Async context manager for lifespan events
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    print('Starting up...')
    db_pool = await create_db_pool()
    redis_client = await create_redis_client()

    # Store in app state
    app.state.db_pool = db_pool
    app.state.redis = redis_client

    yield

    # Shutdown
    print('Shutting down...')
    await db_pool.close()
    await redis_client.close()

app = FastAPI(lifespan=lifespan)

# Custom async context manager
class AsyncDatabaseSession:
    def __init__(self, pool):
        self.pool = pool
        self.conn = None

    async def __aenter__(self):
        self.conn = await self.pool.acquire()
        return self.conn

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await .pool.release(.conn)
         exc_type   :
            
             .conn.rollback()
         


  ():
      AsyncDatabaseSession(app.state.db_pool)  conn:
        result =  conn.fetch()
         result

Connection Pooling

Efficient connection management for databases and HTTP clients.

from fastapi import FastAPI, Depends
import asyncpg
import httpx
from typing import AsyncGenerator

app = FastAPI()

# Database connection pool
class DatabasePool:
    def __init__(self):
        self.pool = None

    async def create_pool(self):
        self.pool = await asyncpg.create_pool(
            'postgresql://user:pass@localhost/db',
            min_size=10,
            max_size=20,
            command_timeout=60,
            max_queries=50000,
            max_inactive_connection_lifetime=300
        )

    async def close_pool(self):
        await self.pool.close()

    async def get_connection(self):
        async with self.pool.acquire() as connection:
            yield connection

db_pool = DatabasePool()

@app.on_event('startup')
async def startup():
    await db_pool.create_pool()


  ():
     db_pool.close_pool()


  ():
    rows =  conn.fetch()
     [(row)  row  rows]


 :
     ():
        .client = 

      () -> AsyncGenerator[httpx.AsyncClient, ]:
         .client  :
            .client = httpx.AsyncClient(
                limits=httpx.Limits(max_keepalive_connections=, max_connections=),
                timeout=httpx.Timeout()
            )
         .client

      ():
         .client:
             .client.aclose()

http_pool = HTTPClientPool()


  ():
    response =  client.get()
     response.json()

Performance Optimization

Async patterns for optimal performance.

from fastapi import FastAPI
import asyncio
from functools import lru_cache

app = FastAPI()

# Cache expensive async operations
from aiocache import Cache
from aiocache.serializers import JsonSerializer

cache = Cache(Cache.MEMORY, serializer=JsonSerializer())

@app.get('/expensive-data/{key}')
async def get_expensive_data(key: str):
    # Check cache first
    cached = await cache.get(key)
    if cached:
        return {'data': cached, 'cached': True}

    # Expensive operation
    await asyncio.sleep(2)
    data = compute_expensive_result(key)

    # Store in cache
    await cache.set(key, data, ttl=300)
    return {'data': data, 'cached': False}

# Batch operations
@app.post('/users/batch')
async def create_users_batch(users: List[UserCreate], db = Depends(get_db)):
    # Create users in batch (more efficient than one-by-one)
    user_objects = [User(**user.dict())  user  users]
    db.add_all(user_objects)
     db.flush()
     user_objects


 :
     ():
        .delay = delay
        .task = 

      ():
         .task:
            .task.cancel()

          ():
             asyncio.sleep(.delay)
             coro

        .task = asyncio.create_task(delayed())
         .task

debouncer = Debouncer(delay=)



  ():
    
    post_task = db.get(Post, post_id)
    comments_task = db.execute(
        select(Comment).where(Comment.post_id == post_id)
    )
    author_task = db.execute(
        select(User).where(User. == Post.author_id)
    )

    post, comments_result, author_result =  asyncio.gather(
        post_task, comments_task, author_task
    )

     {
        : post,
        : comments_result.scalars().(),
        : author_result.scalar_one()
    }

When to Use This Skill

Use fastapi-async-patterns when:

• Building high-throughput APIs that handle many concurrent requests
• Working with I/O-bound operations (database, external APIs, file operations)
• Implementing real-time features (WebSockets, SSE)
• Processing multiple operations in parallel
• Streaming large datasets or files
• Building microservices that communicate with other services
• Optimizing API response times and resource usage
• Handling background tasks without blocking responses

FastAPI Async Best Practices

1. Use async for I/O - Always use async for database, HTTP requests, and file operations
2. Avoid blocking calls - Never use blocking calls in async functions (time.sleep, requests library)
3. Connection pooling - Use connection pools for databases and HTTP clients
4. Proper cleanup - Always clean up resources with try/finally or async context managers
5. Concurrent operations - Use asyncio.gather for parallel operations when possible
6. Background tasks - Use BackgroundTasks for fire-and-forget operations
7. Stream large data - Use StreamingResponse for large files or generated content
8. Timeout handling - Set timeouts on all external calls to prevent hanging
9. Error propagation - Handle exceptions properly in async code
10. Monitor performance - Use tools like aiomonitor to debug async issues

FastAPI Async Common Pitfalls

1. Blocking the event loop - Using synchronous I/O in async functions kills performance
2. Missing await - Forgetting await on async functions causes coroutine warnings
3. Creating too many tasks - Spawning unlimited tasks can exhaust resources
4. Not closing connections - Resource leaks from unclosed database/HTTP connections
5. Mixing sync and async - Incorrect mixing causes event loop issues
6. Race conditions - Shared state in async code without proper locking
7. Timeout issues - No timeouts on external calls can hang the server
8. Memory leaks - Background tasks that never complete accumulate
9. Error swallowing - Silent failures in background tasks and event handlers
10. Deadlocks - Circular waits in async dependencies or locks

Resources

• FastAPI Async Documentation
• Python asyncio Documentation
• SQLAlchemy Async Guide
• HTTPX Async Client
• AsyncPG Documentation
• Motor (MongoDB Async)
• WebSockets in FastAPI
• Server-Sent Events with Starlette