25 FastAPI Interview Questions and Answers for Python Developers
25 most-asked FastAPI interview questions with Python code examples. Covers Pydantic models, dependency injection, middleware, JWT auth, and production deployment.
FastAPI Interview Questions for Freshers
1. What is FastAPI?
FastAPI is a modern, high-performance Python web framework for building APIs. It is built on top of Starlette (for the web parts) and Pydantic (for data validation). FastAPI uses Python type hints to automatically generate API documentation, validate request data, and serialize responses.
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello, FastAPI!"}
2. How is FastAPI different from Flask and Django?
FastAPI is often compared to Flask and Django. Here is a detailed comparison:
| Feature | FastAPI | Flask | Django |
|---|---|---|---|
| Async support | Native (built-in) | Limited (via extensions) | Added in 3.1+ |
| Data validation | Automatic (Pydantic) | Manual | Forms/Serializers |
| Auto documentation | Swagger + ReDoc built-in | No (needs extensions) | No (needs DRF) |
| Performance | Very fast (async) | Moderate | Moderate |
| Type hints | Required, used for validation | Optional | Optional |
| Learning curve | Easy | Easy | Steeper |
3. What are the key features of FastAPI?
- Fast performance: On par with Node.js and Go, thanks to Starlette and async support
- Automatic documentation: Generates Swagger UI and ReDoc from your code
- Data validation: Uses Pydantic models for automatic request/response validation
- Type safety: Leverages Python type hints for editor support and error detection
- Async/await: Native support for asynchronous request handling
- Dependency injection: Built-in DI system for clean, testable code
- Standards-based: Built on OpenAPI and JSON Schema standards
- Security utilities: Built-in OAuth2, JWT, and API key support
4. How do you install and run a FastAPI application?
# Install FastAPI and Uvicorn (ASGI server)
pip install fastapi uvicorn
# Create main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/hello")
async def hello():
return {"message": "Hello World"}
# Run the application
# uvicorn main:app --reload --host 0.0.0.0 --port 8000
5. What is Pydantic and how does FastAPI use it?
Understanding Pydantic is essential for FastAPI. If you are also preparing for general Python interview questions, type hints and data classes are commonly covered there too.
Pydantic is a data validation library that uses Python type annotations. FastAPI uses Pydantic models to automatically validate request bodies, query parameters, and serialize response data.
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
class UserCreate(BaseModel):
name: str = Field(..., min_length=2, max_length=50)
email: EmailStr
age: int = Field(..., ge=18, le=120)
bio: Optional[str] = None
class Config:
json_schema_extra = {
"example": {
"name": "John Doe",
"email": "john@example.com",
"age": 28,
"bio": "Python developer"
}
}
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
# Pydantic automatically validates the request body
return {"id": 1, **user.dict()}
6. What are path parameters and query parameters in FastAPI?
from fastapi import FastAPI, Query
app = FastAPI()
# Path parameter - required, part of URL
@app.get("/users/{user_id}")
async def get_user(user_id: int):
return {"user_id": user_id}
# Query parameters - optional, after ? in URL
@app.get("/items")
async def list_items(
skip: int = Query(default=0, ge=0),
limit: int = Query(default=10, le=100),
search: str = Query(default=None, min_length=2)
):
return {"skip": skip, "limit": limit, "search": search}
# Combined: /products/electronics?page=2&sort=price
@app.get("/products/{category}")
async def get_products(category: str, page: int = 1, sort: str = "name"):
return {"category": category, "page": page, "sort": sort}
7. How do you define a route in FastAPI?
Routes in FastAPI are defined using decorators that correspond to HTTP methods:
from fastapi import FastAPI
app = FastAPI()
@app.get("/items") # GET request
async def read_items():
return [{"id": 1, "name": "Item 1"}]
@app.post("/items") # POST request
async def create_item(item: dict):
return {"created": item}
@app.put("/items/{item_id}") # PUT request
async def update_item(item_id: int, item: dict):
return {"id": item_id, "updated": item}
@app.delete("/items/{item_id}") # DELETE request
async def delete_item(item_id: int):
return {"deleted": item_id}
@app.patch("/items/{item_id}") # PATCH request
async def patch_item(item_id: int, updates: dict):
return {"id": item_id, "patched": updates}
8. What is Dependency Injection in FastAPI?
Dependency Injection (DI) in FastAPI is a system that allows you to declare dependencies that your route functions need. FastAPI automatically resolves and injects them, making code reusable, testable, and clean.
from fastapi import FastAPI, Depends, HTTPException, Header
app = FastAPI()
# Simple dependency
async def get_db():
db = DatabaseSession()
try:
yield db
finally:
db.close()
# Dependency with parameters
async def verify_token(authorization: str = Header(...)):
if authorization != "Bearer valid-token":
raise HTTPException(status_code=401, detail="Invalid token")
return authorization
# Common query params dependency
async def common_params(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
# Using dependencies in routes
@app.get("/users")
async def get_users(
db = Depends(get_db),
params = Depends(common_params),
token = Depends(verify_token)
):
users = db.query(User).offset(params["skip"]).limit(params["limit"]).all()
return users
9. What is the difference between async def and def in FastAPI?
FastAPI supports both synchronous and asynchronous route handlers:
# Async - use for I/O bound operations (DB queries, API calls, file reads)
@app.get("/async-users")
async def get_users_async():
users = await database.fetch_all("SELECT * FROM users")
return users
# Sync - FastAPI runs it in a thread pool automatically
@app.get("/sync-users")
def get_users_sync():
# Blocking I/O - FastAPI handles it in a separate thread
users = requests.get("https://api.example.com/users")
return users.json()
Rule of thumb: Use async def when you can await operations. Use plain def for blocking operations — FastAPI will run them in a thread pool so they don’t block other requests.
10. How does FastAPI handle request validation errors?
FastAPI automatically returns a 422 (Unprocessable Entity) response when validation fails, with details about what went wrong:
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
app = FastAPI()
# Custom error handler for validation errors
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=422,
content={
"detail": exc.errors(),
"message": "Validation failed. Please check your input."
}
)
# If someone sends age="abc" to this endpoint:
@app.get("/users/{user_id}")
async def get_user(user_id: int): # FastAPI validates user_id is an integer
return {"user_id": user_id}
# Response for /users/abc:
# {"detail": [{"loc": ["path", "user_id"], "msg": "value is not a valid integer", "type": "type_error.integer"}]}
FastAPI Interview Questions for Intermediate Developers
11. How do you implement authentication in FastAPI?
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
def create_access_token(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=30))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
async def get_current_user(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise HTTPException(status_code=401, detail="Invalid token")
except JWTError:
raise HTTPException(status_code=401, detail="Invalid token")
return username
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# Verify credentials (simplified)
if form_data.username == "admin" and form_data.password == "secret":
token = create_access_token(data={"sub": form_data.username})
return {"access_token": token, "token_type": "bearer"}
raise HTTPException(status_code=401, detail="Incorrect credentials")
@app.get("/protected")
async def protected_route(current_user: str = Depends(get_current_user)):
return {"message": f"Hello {current_user}"}
12. How do you implement middleware in FastAPI?
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time
app = FastAPI()
# CORS Middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "https://yourdomain.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Custom middleware - request timing
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(round(process_time, 4))
return response
# Custom middleware - logging
@app.middleware("http")
async def log_requests(request: Request, call_next):
print(f"{request.method} {request.url.path}")
response = await call_next(request)
print(f"Status: {response.status_code}")
return response
13. How do you handle file uploads in FastAPI?
from fastapi import FastAPI, File, UploadFile, HTTPException
from typing import List
app = FastAPI()
# Single file upload
@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
if file.content_type not in ["image/jpeg", "image/png"]:
raise HTTPException(status_code=400, detail="Only JPEG/PNG allowed")
contents = await file.read()
file_size = len(contents)
# Save file
with open(f"uploads/{file.filename}", "wb") as f:
f.write(contents)
return {
"filename": file.filename,
"content_type": file.content_type,
"size_kb": round(file_size / 1024, 2)
}
# Multiple file upload
@app.post("/upload-multiple")
async def upload_files(files: List[UploadFile] = File(...)):
results = []
for file in files:
contents = await file.read()
results.append({"filename": file.filename, "size": len(contents)})
return {"uploaded": results}
14. How do you connect FastAPI to a database using SQLAlchemy?
Database integration is a key skill. If you work with NoSQL databases, also check our MongoDB interview questions.
from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
# Database setup
DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Model
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
Base.metadata.create_all(bind=engine)
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
app = FastAPI()
@app.get("/users")
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
users = db.query(User).offset(skip).limit(limit).all()
return users
@app.post("/users")
def create_user(name: str, email: str, db: Session = Depends(get_db)):
user = User(name=name, email=email)
db.add(user)
db.commit()
db.refresh(user)
return user
15. How do you implement background tasks in FastAPI?
from fastapi import FastAPI, BackgroundTasks
import smtplib
app = FastAPI()
def send_email(email: str, message: str):
# Simulating slow email sending
import time
time.sleep(5)
print(f"Email sent to {email}: {message}")
def write_log(message: str):
with open("log.txt", "a") as f:
f.write(f"{message}
")
@app.post("/register")
async def register_user(
email: str,
background_tasks: BackgroundTasks
):
# These run AFTER the response is sent
background_tasks.add_task(send_email, email, "Welcome!")
background_tasks.add_task(write_log, f"New user: {email}")
# Response is returned immediately
return {"message": "User registered. Email will be sent shortly."}
16. How do you use APIRouter for organizing large applications?
# routers/users.py
from fastapi import APIRouter, Depends
router = APIRouter(
prefix="/users",
tags=["users"],
responses={404: {"description": "Not found"}},
)
@router.get("/")
async def list_users():
return [{"id": 1, "name": "Alice"}]
@router.get("/{user_id}")
async def get_user(user_id: int):
return {"id": user_id, "name": "Alice"}
# routers/items.py
from fastapi import APIRouter
router = APIRouter(prefix="/items", tags=["items"])
@router.get("/")
async def list_items():
return [{"id": 1, "name": "Widget"}]
# main.py
from fastapi import FastAPI
from routers import users, items
app = FastAPI()
app.include_router(users.router)
app.include_router(items.router)
17. How do you implement response models and status codes?
from fastapi import FastAPI, status
from pydantic import BaseModel
from typing import Optional, List
app = FastAPI()
class ItemCreate(BaseModel):
name: str
price: float
description: Optional[str] = None
class ItemResponse(BaseModel):
id: int
name: str
price: float
class Config:
from_attributes = True # Pydantic v2
# Explicit response model (hides internal fields)
@app.post("/items", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemCreate):
return {"id": 1, "name": item.name, "price": item.price, "internal_field": "hidden"}
# List response
@app.get("/items", response_model=List[ItemResponse])
async def list_items():
return [{"id": 1, "name": "Widget", "price": 9.99}]
# No content response
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
return None
18. How do you handle WebSockets 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 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: str):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"{client_id}: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"{client_id} left the chat")
FastAPI Interview Questions for Experienced Developers
19. How do you write tests for FastAPI applications?
from fastapi.testclient import TestClient
from main import app
import pytest
client = TestClient(app)
def test_read_root():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello, FastAPI!"}
def test_create_user():
response = client.post("/users", json={
"name": "Alice",
"email": "alice@example.com",
"age": 25
})
assert response.status_code == 201
assert response.json()["name"] == "Alice"
def test_validation_error():
response = client.post("/users", json={
"name": "A", # too short
"email": "invalid-email",
"age": 15 # under 18
})
assert response.status_code == 422
# Async tests with pytest-asyncio
@pytest.mark.asyncio
async def test_async_endpoint():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/async-users")
assert response.status_code == 200
20. How do you implement rate limiting in FastAPI?
from fastapi import FastAPI, Request, HTTPException
from collections import defaultdict
import time
app = FastAPI()
# Simple in-memory rate limiter
rate_limit_store = defaultdict(list)
async def rate_limiter(request: Request, max_requests: int = 10, window: int = 60):
client_ip = request.client.host
now = time.time()
# Remove old entries outside the window
rate_limit_store[client_ip] = [
t for t in rate_limit_store[client_ip] if now - t < window
]
if len(rate_limit_store[client_ip]) >= max_requests:
raise HTTPException(
status_code=429,
detail=f"Rate limit exceeded. Max {max_requests} requests per {window}s."
)
rate_limit_store[client_ip].append(now)
# Using slowapi (production-ready)
# pip install slowapi
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@app.get("/api/data")
@limiter.limit("5/minute")
async def get_data(request: Request):
return {"data": "rate-limited response"}
21. How do you implement caching 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
from redis import asyncio as aioredis
app = FastAPI()
@app.on_event("startup")
async def startup():
redis = aioredis.from_url("redis://localhost:6379")
FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")
# Cache response for 60 seconds
@app.get("/expensive-data")
@cache(expire=60)
async def get_expensive_data():
# Simulate expensive computation
import time
time.sleep(2)
return {"data": "This response is cached for 60 seconds"}
# Manual caching with Redis
@app.get("/users/{user_id}")
async def get_user(user_id: int, redis = Depends(get_redis)):
# Check cache first
cached = await redis.get(f"user:{user_id}")
if cached:
return json.loads(cached)
# Fetch from DB
user = await db.get_user(user_id)
# Store in cache
await redis.setex(f"user:{user_id}", 300, json.dumps(user))
return user
22. How do you deploy FastAPI in production?
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
# Run with Gunicorn + Uvicorn workers
CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]
# docker-compose.yml
version: "3.8"
services:
api:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://user:pass@db:5432/app
depends_on:
- db
- redis
db:
image: postgres:15
environment:
POSTGRES_PASSWORD: pass
redis:
image: redis:7-alpine
For deploying Python applications at scale, understanding containerization is important. Also explore our Django interview questions for comparison on production architecture.
Production checklist:
- Use Gunicorn with Uvicorn workers (multi-process)
- Set up HTTPS via reverse proxy (Nginx/Traefik)
- Enable CORS only for allowed origins
- Use environment variables for secrets
- Add health check endpoint
- Configure structured logging
- Set up monitoring (Prometheus + Grafana)
23. How do you implement event-driven patterns with FastAPI?
from fastapi import FastAPI
from contextlib import asynccontextmanager
# Lifespan events (modern approach, replaces on_event)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup: runs before app starts accepting requests
print("Starting up...")
db_pool = await create_db_pool()
redis = await connect_redis()
app.state.db = db_pool
app.state.redis = redis
yield # App runs here
# Shutdown: runs when app is shutting down
print("Shutting down...")
await db_pool.close()
await redis.close()
app = FastAPI(lifespan=lifespan)
# Startup/shutdown with on_event (older approach)
@app.on_event("startup")
async def startup_event():
app.state.ml_model = load_model("model.pkl")
@app.on_event("shutdown")
async def shutdown_event():
await cleanup_resources()
24. How do you handle pagination in FastAPI?
from fastapi import FastAPI, Query, Depends
from pydantic import BaseModel
from typing import List, Generic, TypeVar
from math import ceil
app = FastAPI()
T = TypeVar("T")
class PaginatedResponse(BaseModel):
items: list
total: int
page: int
page_size: int
total_pages: int
has_next: bool
has_previous: bool
class PaginationParams:
def __init__(
self,
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(10, ge=1, le=100, description="Items per page")
):
self.page = page
self.page_size = page_size
self.offset = (page - 1) * page_size
@app.get("/users", response_model=PaginatedResponse)
async def list_users(
pagination: PaginationParams = Depends(),
db: Session = Depends(get_db)
):
total = db.query(User).count()
users = db.query(User).offset(pagination.offset).limit(pagination.page_size).all()
total_pages = ceil(total / pagination.page_size)
return PaginatedResponse(
items=users,
total=total,
page=pagination.page,
page_size=pagination.page_size,
total_pages=total_pages,
has_next=pagination.page < total_pages,
has_previous=pagination.page > 1
)
25. How do you implement logging and monitoring in FastAPI?
import logging
import sys
from fastapi import FastAPI, Request
import time
import uuid
# Configure structured logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger("fastapi_app")
app = FastAPI()
# Request ID middleware for tracing
@app.middleware("http")
async def add_request_id(request: Request, call_next):
request_id = str(uuid.uuid4())[:8]
start_time = time.time()
logger.info(f"[{request_id}] {request.method} {request.url.path} started")
response = await call_next(request)
duration = round(time.time() - start_time, 4)
logger.info(f"[{request_id}] {request.method} {request.url.path} "
f"completed {response.status_code} in {duration}s")
response.headers["X-Request-ID"] = request_id
return response
# Health check endpoint for monitoring
@app.get("/health")
async def health_check():
return {
"status": "healthy",
"version": "1.0.0",
"timestamp": time.time()
}