FastAPI Interview Questions 56-60 (Scenario-Based: CRUD, Auth, DB, Middleware, Optimization)
Welcome to the final challenge! We have covered the theory, the tools, and the deployment. Now, it is time to put it all together into practical scenarios.
In a real interview, you won't just be asked "What is a dependency?". You will be given a problem statement like "Build a Book Management System" or "Fix this slow API". These five scenarios simulate that experience. They require you to combine multiple concepts (Pydantic, SQLAlchemy, Auth) to solve a business problem.
56. Scenario: Build CRUD APIs for a Book Management System using FastAPI.
This is the most common "take-home" assignment. You need to demonstrate you can structure a basic application with Create, Read, Update, and Delete operations. We will use an in-memory list for simplicity, but the logic applies to databases too.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from uuid import uuid4, UUID
app = FastAPI()
# 1. Define the Data Model
class Book(BaseModel):
id: Optional[UUID] = None
title: str
author: str
pages: int
# In-memory database
books_db: List[Book] = []
# 2. Create (POST)
@app.post("/books/", response_model=Book, status_code=201)
def create_book(book: Book):
book.id = uuid4()
books_db.append(book)
return book
# 3. Read All (GET)
@app.get("/books/", response_model=List[Book])
def get_books():
return books_db
# 4. Read One (GET)
@app.get("/books/{book_id}", response_model=Book)
def get_book(book_id: UUID):
for book in books_db:
if book.id == book_id:
return book
raise HTTPException(status_code=404, detail="Book not found")
# 5. Update (PUT)
@app.put("/books/{book_id}", response_model=Book)
def update_book(book_id: UUID, updated_book: Book):
for i, book in enumerate(books_db):
if book.id == book_id:
# Keep the ID, update other fields
updated_book.id = book_id
books_db[i] = updated_book
return updated_book
raise HTTPException(status_code=404, detail="Book not found")
# 6. Delete (DELETE)
@app.delete("/books/{book_id}", status_code=204)
def delete_book(book_id: UUID):
for i, book in enumerate(books_db):
if book.id == book_id:
books_db.pop(i)
return
raise HTTPException(status_code=404, detail="Book not found")
57. Scenario: Add JWT login/sign-up features to a FastAPI app.
Security is non-negotiable. This scenario tests your ability to handle user credentials safely. We will combine OAuth2PasswordBearer (for token extraction) with passlib (for hashing) and python-jose (for JWT generation).
Steps:
1. User signs up (Hash password -> Save to DB).
2. User logs in (Verify hash -> Generate JWT).
3. User accesses protected route (Verify JWT).
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from passlib.context import CryptContext
from jose import JWTError, jwt
from datetime import datetime, timedelta
# Config
SECRET_KEY = "mysecret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
app = FastAPI()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# Mock User DB
fake_users_db = {}
class User(BaseModel):
username: str
password: str # Plain text (input only)
# Helper: Create Token
def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
# 1. Sign Up
@app.post("/signup")
def signup(user: User):
if user.username in fake_users_db:
raise HTTPException(status_code=400, detail="User already exists")
hashed_password = pwd_context.hash(user.password)
fake_users_db[user.username] = {"username": user.username, "hashed_password": hashed_password}
return {"msg": "User created"}
# 2. Login (Token Generation)
@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = fake_users_db.get(form_data.username)
if not user or not pwd_context.verify(form_data.password, user["hashed_password"]):
raise HTTPException(status_code=400, detail="Incorrect username or password")
access_token = create_access_token(data={"sub": user["username"]})
return {"access_token": access_token, "token_type": "bearer"}
# 3. Protected Route
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)
except JWTError:
raise HTTPException(status_code=401, detail="Invalid credentials")
return username
@app.get("/users/me")
def read_users_me(username: str = Depends(get_current_user)):
return {"username": username}
58. Scenario: Integrate PostgreSQL with SQLAlchemy + Alembic migration.
This scenario proves you can work with real persistent data. You need to setup the engine, create a model, and understand how Alembic fits in for schema changes.
We assume you have a PostgreSQL DB running. This example focuses on the code structure.
# database.py
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# models.py
class UserDB(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
# main.py (Usage)
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
app = FastAPI()
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/users/")
def create_user(email: str, db: Session = Depends(get_db)):
db_user = UserDB(email=email, hashed_password="fakehash")
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
Alembic Commands (Terminal):
# 1. Initialize
alembic init alembic
# 2. Edit alembic.ini with your DB URL
# ...
# 3. Generate migration script
alembic revision --autogenerate -m "Create users table"
# 4. Apply to DB
alembic upgrade head59. Scenario: Create middleware for logging incoming requests and responses.
Observability is key in production. You need to track how long requests take and what status codes are returned. Middleware is the perfect place for this logic because it wraps every request automatically.
import time
import logging
from fastapi import FastAPI, Request
# Configure standard python logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("api_logger")
app = FastAPI()
@app.middleware("http")
async def log_requests(request: Request, call_next):
# 1. Before Request: Record Start Time
start_time = time.time()
# 2. Process Request
response = await call_next(request)
# 3. After Response: Calculate Duration
process_time = time.time() - start_time
# 4. Log details
logger.info(
f"Path: {request.url.path} | "
f"Method: {request.method} | "
f"Status: {response.status_code} | "
f"Duration: {process_time:.4f}s"
)
return response
@app.get("/")
def root():
return {"message": "Hello"}
Sample Output (Terminal):
INFO:api_logger:Path: / | Method: GET | Status: 200 | Duration: 0.0005s60. Scenario: Optimize a slow API endpoint returning large data.
This scenario tests your problem-solving skills. "The API is slow" is a vague complaint. You need to identify the bottleneck (CPU vs IO) and apply the right fix. We will look at optimizing a large JSON response using ORJSON and pagination.
The Problem: Returning 100,000 records at once.
The Solution:
1. Use ORJSONResponse for faster serialization.
2. Implement Pagination (limit/offset).
from fastapi import FastAPI, Query
from fastapi.responses import ORJSONResponse
from typing import List
# Use ORJSON for speed
app = FastAPI(default_response_class=ORJSONResponse)
# Mock huge dataset
fake_big_db = [{"id": i, "name": f"Item {i}"} for i in range(100000)]
@app.get("/items/")
def read_items(
skip: int = Query(0, ge=0), # Offset
limit: int = Query(10, le=1000) # Limit (Max 1000)
):
# Slice the data instead of returning everything
return fake_big_db[skip : skip + limit]
# Why this is faster:
# 1. We send only 10 items, not 100,000.
# 2. ORJSON encodes the JSON response much faster than stdlib json.
This concludes the FastAPI interview preparation course! You have built, secured, deployed, and optimized a modern API. You are now ready to tackle any technical interview with confidence.