FastAPI Testing and Deployment - Interview Questions

1. How do you write unit tests for FastAPI applications?

from fastapi.testclient import TestClient
from fastapi import FastAPI
import pytest
from unittest.mock import Mock, patch

app = FastAPI()

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

@app.get("/users/{user_id}")
def read_user(user_id: int):
    return {"user_id": user_id, "name": "John Doe"}

# Basic test setup
client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"Hello": "World"}

def test_read_user():
    response = client.get("/users/1")
    assert response.status_code == 200
    assert response.json() == {"user_id": 1, "name": "John Doe"}

# Testing with pytest
@pytest.fixture
def client():
    return TestClient(app)

def test_read_root_pytest(client):
    response = client.get("/")
    assert response.status_code == 200
    data = response.json()
    assert data["Hello"] == "World"

# Testing with dependencies
from fastapi import Depends

def get_db():
    return {"users": [{"id": 1, "name": "John"}]}

@app.get("/users")
def get_users(db: dict = Depends(get_db)):
    return db["users"]

def test_get_users():
    with patch("main.get_db") as mock_db:
        mock_db.return_value = {"users": [{"id": 1, "name": "John"}]}
        response = client.get("/users")
        assert response.status_code == 200
        assert response.json() == [{"id": 1, "name": "John"}]

2. How do you test FastAPI endpoints with authentication?

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer
from fastapi.testclient import TestClient
import pytest

app = FastAPI()
security = HTTPBearer()

def get_current_user(token: str = Depends(security)):
    if token != "valid-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )
    return {"user_id": 1, "username": "testuser"}

@app.get("/protected")
def protected_endpoint(current_user: dict = Depends(get_current_user)):
    return {"message": "Protected data", "user": current_user}

client = TestClient(app)

def test_protected_endpoint_with_valid_token():
    headers = {"Authorization": "Bearer valid-token"}
    response = client.get("/protected", headers=headers)
    assert response.status_code == 200
    assert response.json()["message"] == "Protected data"

def test_protected_endpoint_with_invalid_token():
    headers = {"Authorization": "Bearer invalid-token"}
    response = client.get("/protected", headers=headers)
    assert response.status_code == 401

def test_protected_endpoint_without_token():
    response = client.get("/protected")
    assert response.status_code == 403

# Testing with dependency overrides
def get_test_user():
    return {"user_id": 999, "username": "testuser"}

def test_protected_endpoint_with_override():
    app.dependency_overrides[get_current_user] = get_test_user
    response = client.get("/protected")
    assert response.status_code == 200
    assert response.json()["user"]["user_id"] == 999
    app.dependency_overrides.clear()

3. How do you test database operations in FastAPI?

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
from fastapi.testclient import TestClient
import pytest

app = FastAPI()

# Test database setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

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)

def get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

@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 {"id": user.id, "name": user.name, "email": user.email}

@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": user.id, "name": user.name, "email": user.email}

client = TestClient(app)

@pytest.fixture
def db_session():
    Base.metadata.create_all(bind=engine)
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
        Base.metadata.drop_all(bind=engine)

def test_create_user(db_session):
    response = client.post("/users", params={"name": "John", "email": "john@example.com"})
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "John"
    assert data["email"] == "john@example.com"

def test_get_user(db_session):
    # First create a user
    user_response = client.post("/users", params={"name": "Jane", "email": "jane@example.com"})
    user_id = user_response.json()["id"]
    
    # Then get the user
    response = client.get(f"/users/{user_id}")
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Jane"
    assert data["email"] == "jane@example.com"

def test_get_nonexistent_user(db_session):
    response = client.get("/users/999")
    assert response.status_code == 404

4. How do you test async endpoints in FastAPI?

from fastapi import FastAPI
from fastapi.testclient import TestClient
import pytest
import asyncio
import httpx

app = FastAPI()

@app.get("/async-endpoint")
async def async_endpoint():
    await asyncio.sleep(0.1)  # Simulate async operation
    return {"message": "Async response"}

@app.get("/users")
async def get_users():
    # Simulate async database query
    await asyncio.sleep(0.05)
    return [{"id": 1, "name": "John"}, {"id": 2, "name": "Jane"}]

# Testing with TestClient (synchronous)
client = TestClient(app)

def test_async_endpoint():
    response = client.get("/async-endpoint")
    assert response.status_code == 200
    assert response.json()["message"] == "Async response"

def test_get_users():
    response = client.get("/users")
    assert response.status_code == 200
    users = response.json()
    assert len(users) == 2
    assert users[0]["name"] == "John"

# Testing with httpx (async)
@pytest.mark.asyncio
async def test_async_endpoint_with_httpx():
    async with httpx.AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/async-endpoint")
        assert response.status_code == 200
        assert response.json()["message"] == "Async response"

@pytest.mark.asyncio
async def test_get_users_with_httpx():
    async with httpx.AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/users")
        assert response.status_code == 200
        users = response.json()
        assert len(users) == 2

5. How do you set up integration tests for FastAPI?

from fastapi import FastAPI
from fastapi.testclient import TestClient
import pytest
import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Test configuration
TEST_DATABASE_URL = "sqlite:///./test_integration.db"

@pytest.fixture(scope="session")
def test_app():
    # Create test database
    engine = create_engine(TEST_DATABASE_URL)
    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    
    # Create tables
    Base.metadata.create_all(bind=engine)
    
    app = FastAPI()
    
    # Override database dependency
    def get_test_db():
        db = TestingSessionLocal()
        try:
            yield db
        finally:
            db.close()
    
    app.dependency_overrides[get_db] = get_test_db
    
    yield app
    
    # Cleanup
    Base.metadata.drop_all(bind=engine)

@pytest.fixture
def client(test_app):
    return TestClient(test_app)

@pytest.fixture
def db_session(test_app):
    engine = create_engine(TEST_DATABASE_URL)
    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

class TestUserAPI:
    def test_create_and_get_user(self, client, db_session):
        # Create user
        create_response = client.post(
            "/users",
            json={"name": "Test User", "email": "test@example.com"}
        )
        assert create_response.status_code == 200
        user_data = create_response.json()
        
        # Get user
        get_response = client.get(f"/users/{user_data['id']}")
        assert get_response.status_code == 200
        assert get_response.json()["name"] == "Test User"
    
    def test_user_validation(self, client):
        # Test invalid email
        response = client.post(
            "/users",
            json={"name": "Test User", "email": "invalid-email"}
        )
        assert response.status_code == 422
    
    def test_duplicate_email(self, client, db_session):
        # Create first user
        client.post(
            "/users",
            json={"name": "User 1", "email": "duplicate@example.com"}
        )
        
        # Try to create second user with same email
        response = client.post(
            "/users",
            json={"name": "User 2", "email": "duplicate@example.com"}
        )
        assert response.status_code == 400

6. How do you deploy a FastAPI application to production?

# main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    print("Starting up...")
    yield
    # Shutdown
    print("Shutting down...")

app = FastAPI(lifespan=lifespan)

# requirements.txt
"""
fastapi==0.104.1
uvicorn[standard]==0.24.0
gunicorn==21.2.0
python-multipart==0.0.6
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
sqlalchemy==2.0.23
psycopg2-binary==2.9.9
redis==5.0.1
"""

# Dockerfile
"""
FROM python:3.14-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
"""

# docker-compose.yml
"""
version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:password@db:5432/fastapi
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis
  
  db:
    image: postgres:15
    environment:
      - POSTGRES_DB=fastapi
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
    volumes:
      - postgres_data:/var/lib/postgresql/data
  
  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

volumes:
  postgres_data:
  redis_data:
"""

# Production deployment with Gunicorn
# gunicorn.conf.py
"""
bind = "0.0.0.0:8000"
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
max_requests = 1000
max_requests_jitter = 50
timeout = 30
keepalive = 2
preload_app = True
"""

# systemd service file
"""
[Unit]
Description=FastAPI application
After=network.target

[Service]
User=fastapi
Group=fastapi
WorkingDirectory=/opt/fastapi
Environment="PATH=/opt/fastapi/venv/bin"
ExecStart=/opt/fastapi/venv/bin/gunicorn -c gunicorn.conf.py main:app
ExecReload=/bin/kill -s HUP $MAINPID
Restart=always

[Install]
WantedBy=multi-user.target
"""

7. How do you set up CI/CD for FastAPI applications?

# .github/workflows/ci.yml
name: CI/CD Pipeline

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: postgres
          POSTGRES_DB: test_db
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install pytest pytest-asyncio pytest-cov
    
    - name: Run tests
      env:
        DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test_db
      run: |
        pytest --cov=app --cov-report=xml
    
    - name: Upload coverage
      uses: codecov/codecov-action@v3
      with:
        file: ./coverage.xml

  lint:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'
    
    - name: Install dependencies
      run: |
        pip install flake8 black isort mypy
    
    - name: Run linters
      run: |
        flake8 app/
        black --check app/
        isort --check-only app/
        mypy app/

  security:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'
    
    - name: Install dependencies
      run: |
        pip install bandit safety
    
    - name: Run security checks
      run: |
        bandit -r app/
        safety check

  deploy:
    needs: [test, lint, security]
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Deploy to production
      run: |
        echo "Deploying to production..."
        # Add your deployment commands here

8. How do you monitor and log FastAPI applications in production?

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

app = FastAPI()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# Prometheus metrics
REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint', 'status'])
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()
    
    # Log request
    logger.info(f"Request: {request.method} {request.url}")
    
    response = await call_next(request)
    
    # Calculate duration
    duration = time.time() - start_time
    
    # Update metrics
    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=request.url.path,
        status=response.status_code
    ).inc()
    REQUEST_LATENCY.observe(duration)
    
    # Log response
    logger.info(f"Response: {response.status_code} - {duration:.3f}s")
    
    return response

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

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

# Custom logging for specific endpoints
from fastapi import Depends
from pydantic import BaseModel

class UserCreate(BaseModel):
    name: str
    email: str

@app.post("/users")
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
    logger.info(f"Creating user: {user.email}")
    try:
        # User creation logic
        logger.info(f"User created successfully: {user.email}")
        return {"message": "User created"}
    except Exception as e:
        logger.error(f"Error creating user {user.email}: {str(e)}")
        raise

# Structured logging
import json
from datetime import datetime

class StructuredLogger:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    def log_request(self, request: Request, response, duration: float):
        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "method": request.method,
            "url": str(request.url),
            "status_code": response.status_code,
            "duration": duration,
            "user_agent": request.headers.get("user-agent"),
            "ip": request.client.host
        }
        self.logger.info(json.dumps(log_data))

structured_logger = StructuredLogger()

@app.middleware("http")
async def structured_logging(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    duration = time.time() - start_time
    
    structured_logger.log_request(request, response, duration)
    return response

9. How do you handle environment-specific configurations in FastAPI?

from pydantic import BaseSettings
from typing import Optional
import os

class Settings(BaseSettings):
    # Database settings
    database_url: str = "sqlite:///./app.db"
    database_pool_size: int = 10
    database_max_overflow: int = 20
    
    # Security settings
    secret_key: str = "your-secret-key"
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
    
    # Redis settings
    redis_url: str = "redis://localhost:6379"
    
    # External APIs
    external_api_url: str = "https://api.external.com"
    external_api_key: Optional[str] = None
    
    # Logging
    log_level: str = "INFO"
    log_file: str = "app.log"
    
    # Environment
    environment: str = "development"
    debug: bool = False
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

# Environment-specific settings
class DevelopmentSettings(Settings):
    debug: bool = True
    log_level: str = "DEBUG"

class ProductionSettings(Settings):
    debug: bool = False
    log_level: str = "WARNING"
    database_pool_size: int = 20
    database_max_overflow: int = 30

class TestingSettings(Settings):
    database_url: str = "sqlite:///./test.db"
    debug: bool = True
    log_level: str = "DEBUG"

def get_settings() -> Settings:
    environment = os.getenv("ENVIRONMENT", "development")
    
    if environment == "production":
        return ProductionSettings()
    elif environment == "testing":
        return TestingSettings()
    else:
        return DevelopmentSettings()

settings = get_settings()

# Using settings in the application
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI(debug=settings.debug)

@app.get("/config")
async def get_config():
    return {
        "environment": settings.environment,
        "debug": settings.debug,
        "database_url": settings.database_url
    }

# Environment-specific database setup
def get_database_url():
    if settings.environment == "testing":
        return "sqlite:///./test.db"
    elif settings.environment == "production":
        return os.getenv("DATABASE_URL")
    else:
        return "sqlite:///./dev.db"

# Environment-specific middleware
if settings.environment == "production":
    from fastapi.middleware.trustedhost import TrustedHostMiddleware
    app.add_middleware(
        TrustedHostMiddleware,
        allowed_hosts=["yourdomain.com", "*.yourdomain.com"]
    )

10. How do you implement blue-green deployment for FastAPI?

# Blue-green deployment configuration
# docker-compose.blue.yml
"""
version: '3.8'

services:
  web-blue:
    build: .
    image: fastapi-app:blue
    ports:
      - "8001:8000"
    environment:
      - ENVIRONMENT=production
      - VERSION=blue
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  web-green:
    build: .
    image: fastapi-app:green
    ports:
      - "8002:8000"
    environment:
      - ENVIRONMENT=production
      - VERSION=green
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - web-blue
      - web-green
"""

# nginx.conf for load balancing
"""
events {
    worker_connections 1024;
}

http {
    upstream backend {
        server web-blue:8000;
        server web-green:8000 backup;
    }

    server {
        listen 80;
        
        location / {
            proxy_pass http://backend;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
        }
        
        location /health {
            proxy_pass http://backend;
        }
    }
}
"""

# Deployment script
"""
#!/bin/bash

# Blue-green deployment script

set -e

# Configuration
BLUE_PORT=8001
GREEN_PORT=8002
NGINX_PORT=80
HEALTH_CHECK_URL="http://localhost"

# Function to check if service is healthy
check_health() {
    local port=$1
    local max_attempts=30
    local attempt=1
    
    while [ $attempt -le $max_attempts ]; do
        if curl -f "$HEALTH_CHECK_URL:$port/health" > /dev/null 2>&1; then
            echo "Service on port $port is healthy"
            return 0
        fi
        
        echo "Attempt $attempt: Service on port $port is not ready yet..."
        sleep 10
        attempt=$((attempt + 1))
    done
    
    echo "Service on port $port failed health check"
    return 1
}

# Function to switch traffic
switch_traffic() {
    local active_port=$1
    local new_port=$2
    
    echo "Switching traffic from port $active_port to port $new_port"
    
    # Update nginx configuration
    sed -i "s/server web-$active_port:8000;/server web-$active_port:8000 backup;/g" nginx.conf
    sed -i "s/server web-$new_port:8000 backup;/server web-$new_port:8000;/g" nginx.conf
    
    # Reload nginx
    docker-compose exec nginx nginx -s reload
    
    echo "Traffic switched successfully"
}

# Main deployment logic
echo "Starting blue-green deployment..."

# Determine current active environment
if curl -f "$HEALTH_CHECK_URL:$BLUE_PORT/health" > /dev/null 2>&1; then
    CURRENT_ACTIVE="blue"
    NEW_ENVIRONMENT="green"
    CURRENT_PORT=$BLUE_PORT
    NEW_PORT=$GREEN_PORT
elif curl -f "$HEALTH_CHECK_URL:$GREEN_PORT/health" > /dev/null 2>&1; then
    CURRENT_ACTIVE="green"
    NEW_ENVIRONMENT="blue"
    CURRENT_PORT=$GREEN_PORT
    NEW_PORT=$BLUE_PORT
else
    echo "No active environment found, starting with blue"
    CURRENT_ACTIVE="none"
    NEW_ENVIRONMENT="blue"
    NEW_PORT=$BLUE_PORT
fi

echo "Current active: $CURRENT_ACTIVE"
echo "New environment: $NEW_ENVIRONMENT"

# Deploy new environment
echo "Deploying new environment ($NEW_ENVIRONMENT)..."
docker-compose up -d web-$NEW_ENVIRONMENT

# Wait for new environment to be healthy
echo "Waiting for new environment to be healthy..."
if ! check_health $NEW_PORT; then
    echo "New environment failed health check, rolling back..."
    docker-compose stop web-$NEW_ENVIRONMENT
    exit 1
fi

# Switch traffic if there was a previous active environment
if [ "$CURRENT_ACTIVE" != "none" ]; then
    switch_traffic $CURRENT_ACTIVE $NEW_ENVIRONMENT
    
    # Wait a bit to ensure traffic is stable
    sleep 30
    
    # Stop old environment
    echo "Stopping old environment ($CURRENT_ACTIVE)..."
    docker-compose stop web-$CURRENT_ACTIVE
else
    # First deployment, just start nginx
    echo "First deployment, starting nginx..."
    docker-compose up -d nginx
fi

echo "Blue-green deployment completed successfully!"
"""

## Interview angle

- **"How do you test a FastAPI app?"** - `TestClient` for sync tests, `httpx.AsyncClient` with ASGI transport for async ones, and `app.dependency_overrides` to swap the database and external clients for test doubles. Real dependencies come from Testcontainers at the integration layer.
- **"How do you deploy it?"** - an ASGI server (uvicorn, or gunicorn with uvicorn workers) behind a reverse proxy for TLS, static files and buffering. Worker count from cores; note that each worker is a separate process with its own connection pool, which is a common source of pool exhaustion.
- **"What belongs in the lifespan handler?"** - creating and disposing shared resources: connection pools, HTTP clients, caches. Creating an HTTP client per request throws away connection reuse and is a frequent performance bug.
- **"How do you handle migrations on deploy?"** - a separate migration step, not on application startup, and never concurrently from every replica. Use a job or an advisory lock so exactly one runs. See [../../08_databases/sql/16_alembic.md](../../08_databases/sql/16_alembic.md).