FastAPI Async Programming and Performance - Interview Questions

1. What is async programming in FastAPI and why is it important?

Async programming in FastAPI allows handling multiple requests concurrently without blocking. It’s important because:

• Better Performance: Can handle many concurrent requests efficiently
• Non-blocking I/O: Database queries, HTTP requests don’t block the server
• Scalability: Better resource utilization
• Responsiveness: Server remains responsive during I/O operations

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "Hello World"}

@app.get("/users")
async def get_users():
    # Simulate async database query
    await asyncio.sleep(1)
    return {"users": ["user1", "user2"]}

2. What’s the difference between sync and async functions in FastAPI?

Sync Functions:

@app.get("/sync")
def sync_endpoint():
    # This blocks the entire thread
    time.sleep(1)
    return {"message": "sync"}

Async Functions:

@app.get("/async")
async def async_endpoint():
    # This doesn't block, allows other requests
    await asyncio.sleep(1)
    return {"message": "async"}

Key Differences:

• Sync functions block the thread during I/O
• Async functions yield control during I/O operations
• Async functions can handle more concurrent requests
• Sync functions are simpler but less scalable

3. How do you handle async database operations in FastAPI?

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
import asyncio

app = FastAPI()

# Async database setup
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_async_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute("SELECT * FROM users")
    users = result.fetchall()
    return {"users": users}

@app.post("/users")
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_async_db)):
    db_user = User(**user.dict())
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user

4. How do you make HTTP requests asynchronously in FastAPI?

from fastapi import FastAPI
import httpx
import asyncio

app = FastAPI()

@app.get("/external-data")
async def get_external_data():
    async with httpx.AsyncClient() as client:
        # Make multiple requests concurrently
        responses = await asyncio.gather(
            client.get("https://api1.com/data"),
            client.get("https://api2.com/data"),
            client.get("https://api3.com/data")
        )
        
        data = [response.json() for response in responses]
        return {"data": data}

# Alternative with aiohttp
import aiohttp

@app.get("/external-data-aiohttp")
async def get_external_data_aiohttp():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/data") as response:
            data = await response.json()
            return data

5. How do you handle background tasks in FastAPI?

from fastapi import FastAPI, BackgroundTasks
import asyncio

app = FastAPI()

def send_email(email: str, message: str):
    # Simulate sending email
    print(f"Sending email to {email}: {message}")

async def process_data_async(data: dict):
    # Simulate async processing
    await asyncio.sleep(5)
    print(f"Processed data: {data}")

@app.post("/users")
async def create_user(user: UserCreate, background_tasks: BackgroundTasks):
    # Add background tasks
    background_tasks.add_task(send_email, user.email, "Welcome!")
    background_tasks.add_task(process_data_async, user.dict())
    
    return {"message": "User created", "user": user}

# Using asyncio.create_task for more control
@app.post("/users-advanced")
async def create_user_advanced(user: UserCreate):
    # Create background task
    task = asyncio.create_task(process_data_async(user.dict()))
    
    return {"message": "User created", "task_id": id(task)}

6. What are the performance optimization techniques in FastAPI?

1. Use Async Functions:

@app.get("/users")
async def get_users():  # Better than sync
    await db.fetch_users()
    return users

2. Connection Pooling:

from databases import Database

database = Database("postgresql://user:pass@localhost/db", min_size=5, max_size=20)

3. Caching:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
from redis import asyncio as aioredis

@asynccontextmanager
async def lifespan(app: FastAPI):
    redis = aioredis.from_url("redis://localhost", encoding="utf8")
    FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")
    yield
    await redis.close()

app = FastAPI(lifespan=lifespan)

@app.get("/users")
@cache(expire=60)
async def get_users():
    return await db.fetch_users()

Note: @app.on_event("startup") / @app.on_event("shutdown") are deprecated. Use the lifespan async-context-manager pattern shown above — both ends of the app lifecycle live in one function, with yield separating startup from shutdown.

4. Response Streaming:

from fastapi.responses import StreamingResponse

@app.get("/large-file")
async def get_large_file():
    def generate():
        for i in range(1000000):
            yield f"Line {i}\n"
    
    return StreamingResponse(generate(), media_type="text/plain")

7. How do you handle concurrent database operations?

from fastapi import FastAPI
import asyncio
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

@app.get("/users-stats")
async def get_user_stats(db: AsyncSession = Depends(get_async_db)):
    # Execute multiple queries concurrently
    tasks = [
        db.execute("SELECT COUNT(*) FROM users"),
        db.execute("SELECT COUNT(*) FROM users WHERE active = true"),
        db.execute("SELECT AVG(age) FROM users")
    ]
    
    results = await asyncio.gather(*tasks)
    
    return {
        "total_users": results[0].scalar(),
        "active_users": results[1].scalar(),
        "avg_age": results[2].scalar()
    }

# Using connection pooling for multiple databases
async def get_user_data(user_id: int):
    async with AsyncSessionLocal() as db:
        user = await db.get(User, user_id)
        return user

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await get_user_data(user_id)
    return user

8. How do you implement rate limiting in FastAPI?

from fastapi import FastAPI, HTTPException, Depends
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

app = FastAPI()
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/api/data")
@limiter.limit("5/minute")
async def get_data(request: Request):
    return {"data": "some data"}

# Custom rate limiting
from collections import defaultdict
import time

class RateLimiter:
    def __init__(self, requests_per_minute: int = 60):
        self.requests_per_minute = requests_per_minute
        self.requests = defaultdict(list)
    
    def is_allowed(self, client_id: str) -> bool:
        now = time.time()
        minute_ago = now - 60
        
        # Clean old requests
        self.requests[client_id] = [
            req_time for req_time in self.requests[client_id]
            if req_time > minute_ago
        ]
        
        if len(self.requests[client_id]) >= self.requests_per_minute:
            return False
        
        self.requests[client_id].append(now)
        return True

rate_limiter = RateLimiter()

@app.get("/protected")
async def protected_endpoint(client_id: str = Header(...)):
    if not rate_limiter.is_allowed(client_id):
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
    return {"message": "Success"}

9. How do you handle WebSocket connections in FastAPI?

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    
    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)
    
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_personal_message(f"You wrote: {data}", websocket)
            await manager.broadcast(f"Client #{client_id} says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left the chat")

10. How do you implement caching strategies in FastAPI?

from fastapi import FastAPI, Depends
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
import aioredis

app = FastAPI()

@app.on_event("startup")
async def startup():
    redis = aioredis.from_url("redis://localhost", encoding="utf8")
    FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")

# Simple caching
@app.get("/users")
@cache(expire=60)  # Cache for 60 seconds
async def get_users():
    return await db.fetch_users()

# Conditional caching
@app.get("/user/{user_id}")
@cache(expire=300, key_builder=lambda func, *args, **kwargs: f"user:{kwargs['user_id']}")
async def get_user(user_id: int):
    return await db.fetch_user(user_id)

# Manual cache management
@app.get("/expensive-data")
async def get_expensive_data():
    cache_key = "expensive_data"
    
    # Try to get from cache
    cached_data = await FastAPICache.get(cache_key)
    if cached_data:
        return cached_data
    
    # Compute expensive data
    data = await compute_expensive_data()
    
    # Store in cache
    await FastAPICache.set(cache_key, data, expire=3600)
    return data

11. How do you monitor and profile FastAPI applications?

from fastapi import FastAPI, Request
import time
import logging
from prometheus_client import Counter, Histogram, generate_latest

app = FastAPI()

# Metrics
REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint'])
REQUEST_LATENCY = Histogram('http_request_duration_seconds', 'HTTP request latency')

# Middleware for monitoring
@app.middleware("http")
async def monitor_requests(request: Request, call_next):
    start_time = time.time()
    
    response = await call_next(request)
    
    duration = time.time() - start_time
    REQUEST_COUNT.labels(method=request.method, endpoint=request.url.path).inc()
    REQUEST_LATENCY.observe(duration)
    
    return response

# Health check endpoint
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

# Metrics endpoint
@app.get("/metrics")
async def metrics():
    return Response(generate_latest(), media_type="text/plain")

# Custom logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    logger.info(f"Response: {response.status_code}")
    return response

12. How do you handle database connection pooling and optimization?

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool

app = FastAPI()

# Optimized database configuration
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,  # Number of connections to maintain
    max_overflow=30,  # Additional connections when pool is full
    pool_pre_ping=True,  # Verify connections before use
    pool_recycle=3600,  # Recycle connections after 1 hour
    echo=False  # Set to True for SQL logging
)

AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False
)

async def get_async_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

# Using connection pooling efficiently
@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_async_db)):
    # Use connection from pool
    result = await db.execute("SELECT * FROM users LIMIT 100")
    users = result.fetchall()
    return {"users": users}

Interview angle

• “def or async def for a route?” - async def when the body awaits async I/O. Plain def when it does blocking work, because FastAPI runs those in a threadpool automatically. The dangerous combination is async def containing a blocking call, which stalls the whole event loop.
• “How would you diagnose a blocked event loop?” - p99 latency rises across every endpoint simultaneously, including ones doing no work. Find the sync call - a sync DB driver, requests, file I/O, or CPU work - and move it to asyncio.to_thread or a process pool.
• “Does async make it faster?” - it increases concurrency for I/O-bound work, not raw speed. CPU-bound endpoints get no benefit and actively harm other requests if run on the loop.
• “What else moves the needle?” - connection pooling with a properly sized async pool, avoiding N+1 queries, response model size, and gzip for large payloads. Most latency problems are the database, not the framework.