Building with FastAPI: A Production Journey (Real Code, Real Lessons)

How I built high-performance microservices for handling 1,000+ requests per second

---

The FastAPI Choice

When I started VehicleMetrics, I could have chosen Django, Flask, or even Go.

But I chose FastAPI.

Not because it's trendy. Not because of hype. But because it solves real problems that matter at scale.

Let me show you why, with actual code from production.

---

Why FastAPI Over Django/Flask?

Django

Pros: Batteries included. ORM. Admin panel. Massive ecosystem.
Cons: Bloated for microservices. Slow request handling. Heavyweight framework for what we need.

Flask

Pros: Lightweight. Simple. Easy to start.
Cons: No built-in validation. No async support by default. Manual everything.

FastAPI

Pros:

• Built-in async/await (critical for I/O-bound work)
• Automatic OpenAPI documentation
• Runtime validation with Pydantic
• Dependency injection (clean, testable code)
• 10-100x faster than Flask/Django
• Type hints = self-documenting

Decision: FastAPI. It's the sweet spot between simplicity and power.

---

The Ingestion Service: 1,000 RPS Challenge

Here's the real challenge: receive and process 1,000 sensor updates per second, with < 50ms response time.

The Naive Approach (That Fails)

from fastapi import FastAPI
from sqlalchemy import create_engine

app = FastAPI()
engine = create_engine("postgresql://...")
connection = engine.connect()

@app.post("/api/vehicles/{vehicle_id}/sensor-data")
def ingest_data(vehicle_id: str, data: dict):
    # ❌ BLOCKING: Waits for database
    result = connection.execute(
        "INSERT INTO sensor_data ..."
    )
    return {"status": "saved"}

Problems:

• Database insert is synchronous (blocks the thread)
• 1,000 requests = need 1,000 threads waiting on I/O
• Thread overhead kills performance
• Response time: 200ms+ (database latency)

The Production Approach (What Works)

from fastapi import FastAPI, Depends, BackgroundTasks
from pydantic import BaseModel, Field, validator
from typing import Optional
import aiokafka
import redis.asyncio as redis
import logging

logger = logging.getLogger(__name__)

# 1. Define your data model with validation
class SensorData(BaseModel):
    vehicle_id: str = Field(..., min_length=1, max_length=50)
    timestamp: int = Field(..., gt=0)
    speed: float = Field(..., ge=0, le=200)
    lat: float = Field(..., ge=-90, le=90)
    lon: float = Field(..., ge=-180, le=180)
    sensors: dict = Field(default={})

    @validator('timestamp')
    def validate_timestamp(cls, v):
        import time
        current = int(time.time() * 1000)
        # Reject data older than 5 minutes
        if abs(current - v) > 300000:
            raise ValueError('Timestamp too old or in future')
        return v

# 2. Initialize async clients (connection pooling)
class AppState:
    kafka: Optional[aiokafka.AIOKafkaProducer] = None
    redis: Optional[redis.Redis] = None

app = FastAPI()
state = AppState()

@app.on_event("startup")
async def startup():
    # Kafka connection pool
    state.kafka = aiokafka.AIOKafkaProducer(
        bootstrap_servers='kafka:9092',
        compression_type='gzip',
        batch_size=16384,
    )
    await state.kafka.start()

    # Redis connection pool
    state.redis = await redis.from_url(
        "redis://localhost:6379",
        encoding="utf8",
        decode_responses=True,
        max_connections=50,
    )
    logger.info("Startup: Kafka and Redis initialized")

@app.on_event("shutdown")
async def shutdown():
    if state.kafka:
        await state.kafka.stop()
    if state.redis:
        await state.redis.close()
    logger.info("Shutdown: Connections closed")

# 3. The actual endpoint
@app.post("/api/vehicles/{vehicle_id}/sensor-data", status_code=201)
async def ingest_sensor_data(
    vehicle_id: str,
    data: SensorData,
    background_tasks: BackgroundTasks,
):
    """
    Ingest vehicle sensor data.

    - Validates payload (Pydantic)
    - De-duplicates (Redis check)
    - Sends to Kafka (fire-and-forget)
    - Returns immediately (< 50ms)
    """

    try:
        # 1. De-duplication check (Redis, async)
        cache_key = f"vehicle:{vehicle_id}:latest:{data.timestamp}"
        is_duplicate = await state.redis.exists(cache_key)

        if is_duplicate:
            logger.debug(f"Duplicate message for {vehicle_id}")
            return {
                "status": "ignored",
                "reason": "duplicate",
                "vehicle_id": vehicle_id,
            }

        # 2. Mark as seen (Redis, 10-second TTL)
        await state.redis.setex(
            cache_key,
            ex=10,
            value="1"
        )

        # 3. Update latest vehicle state (for dashboard)
        latest_key = f"vehicle:{vehicle_id}:latest"
        await state.redis.hset(
            latest_key,
            mapping={
                "speed": data.speed,
                "lat": data.lat,
                "lon": data.lon,
                "timestamp": data.timestamp,
            }
        )
        await state.redis.expire(latest_key, 3600)  # 1 hour TTL

        # 4. Send to Kafka (async, non-blocking)
        message = {
            "vehicle_id": vehicle_id,
            "timestamp": data.timestamp,
            "speed": data.speed,
            "lat": data.lat,
            "lon": data.lon,
            "sensors": data.sensors,
        }

        # Don't await - fire and forget
        background_tasks.add_task(
            state.kafka.send_and_wait,
            "sensor-data",
            value=json.dumps(message).encode(),
            key=vehicle_id.encode(),
            partition=hash(vehicle_id) % 6,  # Route to specific partition
        )

        logger.info(
            f"Ingested {vehicle_id} - speed: {data.speed}",
            extra={"vehicle_id": vehicle_id, "timestamp": data.timestamp}
        )

        return {
            "status": "accepted",
            "vehicle_id": vehicle_id,
            "timestamp": data.timestamp,
        }

    except Exception as e:
        logger.error(f"Ingest error: {str(e)}", exc_info=True)
        raise HTTPException(status_code=500, detail="Ingest failed")

Why this works:

1. Pydantic validation: Rejects bad data instantly (type-safe)
2. Async I/O: No threads blocking on I/O
3. Connection pooling: Reuse connections, not create new ones
4. Fire-and-forget: Kafka send doesn't block response
5. Redis caching: Dedup check in < 2ms (memory)
6. Response time: < 50ms (tested)

Performance:

• 1 instance handles 100+ rps
• 10 instances = 1,000+ rps
• Each handles concurrent requests with same thread

---

Dependency Injection: The Secret Weapon

FastAPI's dependency system is game-changing. Here's why:

Without DI (Messy)

@app.get("/api/vehicle/{id}")
async def get_vehicle(id: str):
    # Auth check
    user = await check_auth(request.headers["Authorization"])

    # Database query
    db = get_database()
    vehicle = await db.fetch_one(...)

    # Authorization check
    if vehicle.owner_id != user.id:
        raise PermissionError()

    return vehicle

Problems:

• Mixed concerns
• Hard to test
• Duplicate auth code everywhere
• Tight coupling

With DI (Clean)

from fastapi import Depends, HTTPException, Request
from functools import lru_cache

# 1. Define dependencies
async def get_current_user(
    request: Request,
    redis: redis.Redis = Depends(get_redis_client),
) -> User:
    """Extract and validate JWT from Authorization header."""
    token = request.headers.get("Authorization", "").replace("Bearer ", "")

    if not token:
        raise HTTPException(status_code=401, detail="Missing token")

    # Check Redis cache first (avoid JWT parsing)
    cached = await redis.get(f"user_token:{token}")
    if cached:
        return User(**json.loads(cached))

    # Validate JWT
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user = User(**payload)

        # Cache for 24 hours
        await redis.setex(
            f"user_token:{token}",
            ex=86400,
            value=json.dumps(user.dict())
        )
        return user
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

async def get_vehicle_or_404(
    vehicle_id: str,
    db = Depends(get_database),
) -> Vehicle:
    """Fetch vehicle or raise 404."""
    vehicle = await db.fetch_one(
        "SELECT * FROM vehicles WHERE id = $1",
        vehicle_id
    )
    if not vehicle:
        raise HTTPException(status_code=404, detail="Vehicle not found")
    return Vehicle(**vehicle)

async def check_vehicle_ownership(
    current_user: User = Depends(get_current_user),
    vehicle: Vehicle = Depends(get_vehicle_or_404),
) -> Vehicle:
    """Verify user owns this vehicle."""
    if vehicle.owner_id != current_user.id:
        raise HTTPException(status_code=403, detail="Not authorized")
    return vehicle

# 2. Use dependencies (clean!)
@app.get("/api/vehicles/{vehicle_id}")
async def get_vehicle_details(
    vehicle: Vehicle = Depends(check_vehicle_ownership),
) -> dict:
    """Get vehicle data. Authentication + authorization automatic."""
    return {
        "id": vehicle.id,
        "name": vehicle.name,
        "status": vehicle.status,
    }

@app.put("/api/vehicles/{vehicle_id}")
async def update_vehicle(
    vehicle: Vehicle = Depends(check_vehicle_ownership),
    update: VehicleUpdate,
    db = Depends(get_database),
) -> Vehicle:
    """Update vehicle. Auth included."""
    await db.execute(
        "UPDATE vehicles SET name = $1 WHERE id = $2",
        update.name,
        vehicle.id,
    )
    return await db.fetch_one("SELECT * FROM vehicles WHERE id = $1", vehicle.id)

Benefits:

• Separation of concerns
• Reusable dependencies
• Easy testing (mock dependencies)
• Clear what each endpoint needs
• DRY (Don't Repeat Yourself)

---

Error Handling: Don't Let Exceptions Escape

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
import logging

logger = logging.getLogger(__name__)
app = FastAPI()

# 1. Validation errors (Pydantic)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Return clean validation errors."""
    return JSONResponse(
        status_code=422,
        content={
            "status": "validation_error",
            "errors": [
                {
                    "field": error["loc"][-1],
                    "message": error["msg"],
                    "type": error["type"],
                }
                for error in exc.errors()
            ]
        }
    )

# 2. Application errors (Custom)
class VehicleNotFound(Exception):
    def __init__(self, vehicle_id: str):
        self.vehicle_id = vehicle_id
        self.message = f"Vehicle {vehicle_id} not found"

@app.exception_handler(VehicleNotFound)
async def vehicle_not_found_handler(request: Request, exc: VehicleNotFound):
    """Handle missing vehicle gracefully."""
    logger.warning(f"Vehicle not found: {exc.vehicle_id}")
    return JSONResponse(
        status_code=404,
        content={"status": "error", "message": exc.message}
    )

# 3. Catch-all (unexpected errors)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    """Log unexpected errors, return safe response."""
    logger.error(
        f"Unexpected error: {str(exc)}",
        exc_info=True,
        extra={"path": request.url.path}
    )

    # Don't leak internal details in production
    return JSONResponse(
        status_code=500,
        content={"status": "error", "message": "Internal server error"}
    )

# Usage
@app.get("/api/vehicles/{vehicle_id}")
async def get_vehicle(vehicle_id: str, db = Depends(get_database)):
    vehicle = await db.fetch_one(
        "SELECT * FROM vehicles WHERE id = $1",
        vehicle_id
    )

    if not vehicle:
        raise VehicleNotFound(vehicle_id)  # Clean, handled above

    return vehicle

---

Request/Response Logging (Without Slowing Down)

import time
import uuid
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from contextvars import ContextVar

# Store request ID in context for logging
request_id_var: ContextVar[str] = ContextVar("request_id", default="")

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 1. Generate request ID
        request_id = str(uuid.uuid4())
        request_id_var.set(request_id)

        # 2. Log request start
        logger.info(
            f"→ {request.method} {request.url.path}",
            extra={
                "request_id": request_id,
                "method": request.method,
                "path": request.url.path,
                "client": request.client.host if request.client else "unknown",
            }
        )

        # 3. Time the request
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start

        # 4. Log response
        logger.info(
            f"← {response.status_code} ({duration*1000:.1f}ms)",
            extra={
                "request_id": request_id,
                "status_code": response.status_code,
                "duration_ms": duration * 1000,
            }
        )

        # 5. Add request ID to response headers (for debugging)
        response.headers["X-Request-ID"] = request_id

        return response

app.add_middleware(LoggingMiddleware)

Output:

2026-07-15 10:23:45 INFO → POST /api/vehicles/V123/sensor-data (request_id=abc-123)
2026-07-15 10:23:45 INFO ← 201 (12.3ms) (request_id=abc-123)

---

Database Queries (The Right Way)

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

# 1. Create async engine
engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/vehicle_metrics",
    echo=False,  # Set to True for SQL logging
    pool_size=20,
    max_overflow=10,
)

# 2. Session factory
async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_database():
    async with async_session() as session:
        yield session

# 3. Parameterized queries (prevent SQL injection)
@app.get("/api/vehicles/{vehicle_id}/analytics")
async def get_vehicle_analytics(
    vehicle_id: str,
    days: int = 30,
    db: AsyncSession = Depends(get_database),
):
    """Get vehicle analytics for last N days."""

    # Use parameters, not string concatenation
    query = text("""
        SELECT 
            DATE_TRUNC('hour', timestamp) as hour,
            AVG(speed) as avg_speed,
            MAX(speed) as max_speed,
            COUNT(*) as samples
        FROM sensor_data
        WHERE vehicle_id = :vehicle_id
          AND timestamp > NOW() - INTERVAL ':days days'
        GROUP BY hour
        ORDER BY hour DESC
    """)

    result = await db.execute(
        query,
        {
            "vehicle_id": vehicle_id,
            "days": days,
        }
    )

    return [dict(row) for row in result.fetchall()]

# 4. Transactions (for consistency)
@app.post("/api/vehicles/{vehicle_id}/alert")
async def create_alert(
    vehicle_id: str,
    alert: AlertCreate,
    db: AsyncSession = Depends(get_database),
):
    """Create alert + update vehicle status (atomic)."""

    try:
        # Both queries run in a transaction
        await db.execute(
            text("INSERT INTO alerts (vehicle_id, message) VALUES (:vehicle_id, :message)"),
            {"vehicle_id": vehicle_id, "message": alert.message}
        )

        await db.execute(
            text("UPDATE vehicles SET status = 'alert' WHERE id = :vehicle_id"),
            {"vehicle_id": vehicle_id}
        )

        await db.commit()  # Commit both at once
        return {"status": "created"}

    except Exception as e:
        await db.rollback()  # Rollback both if error
        logger.error(f"Alert creation failed: {str(e)}")
        raise

---

Testing FastAPI (So Much Easier)

from fastapi.testclient import TestClient
import pytest

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

def test_ingest_valid_data(client):
    """Test successful data ingestion."""
    response = client.post(
        "/api/vehicles/V123/sensor-data",
        json={
            "timestamp": 1000000,
            "speed": 55.5,
            "lat": 37.7749,
            "lon": -122.4194,
            "sensors": {}
        }
    )

    assert response.status_code == 201
    assert response.json()["status"] == "accepted"

def test_ingest_invalid_speed(client):
    """Test validation rejects invalid speed."""
    response = client.post(
        "/api/vehicles/V123/sensor-data",
        json={
            "timestamp": 1000000,
            "speed": 999,  # ❌ Exceeds max of 200
            "lat": 37.7749,
            "lon": -122.4194,
            "sensors": {}
        }
    )

    assert response.status_code == 422
    assert "speed" in str(response.json())

def test_ingest_missing_vehicle_id(client):
    """Test validation requires vehicle_id."""
    response = client.post(
        "/api//sensor-data",  # Missing vehicle_id
        json={
            "timestamp": 1000000,
            "speed": 55.5,
            "lat": 37.7749,
            "lon": -122.4194,
            "sensors": {}
        }
    )

    assert response.status_code == 404

---

Configuration Management

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    # Database
    database_url: str = "postgresql://localhost/vehicle_metrics"
    database_pool_size: int = 20

    # Redis
    redis_url: str = "redis://localhost:6379"
    redis_max_connections: int = 50

    # Kafka
    kafka_bootstrap_servers: str = "localhost:9092"
    kafka_topic: str = "sensor-data"

    # API
    api_title: str = "VehicleMetrics API"
    api_version: str = "1.0.0"
    debug: bool = False

    # Security
    secret_key: str = "change-me-in-production"
    jwt_algorithm: str = "HS256"
    jwt_expiration_hours: int = 24

    class Config:
        env_file = ".env"  # Read from .env file
        case_sensitive = False

@lru_cache()
def get_settings():
    return Settings()

# Usage
@app.get("/api/health")
async def health_check(settings: Settings = Depends(get_settings)):
    return {
        "status": "healthy",
        "version": settings.api_version,
    }

.env file:

DATABASE_URL=postgresql://user:pass@localhost/vehicle_metrics
REDIS_URL=redis://localhost:6379
KAFKA_BOOTSTRAP_SERVERS=kafka:9092
SECRET_KEY=your-super-secret-key
DEBUG=False

---

Monitoring & Metrics

from prometheus_client import Counter, Histogram, generate_latest
import time

# Metrics
request_count = Counter(
    'fastapi_requests_total',
    'Total requests',
    ['method', 'endpoint', 'status']
)

request_duration = Histogram(
    'fastapi_request_duration_seconds',
    'Request duration',
    ['method', 'endpoint'],
    buckets=(0.01, 0.05, 0.1, 0.5, 1.0)
)

@app.middleware("http")
async def add_metrics(request: Request, call_next):
    start = time.time()
    response = await call_next(request)

    duration = time.time() - start
    request_count.labels(
        method=request.method,
        endpoint=request.url.path,
        status=response.status_code,
    ).inc()

    request_duration.labels(
        method=request.method,
        endpoint=request.url.path,
    ).observe(duration)

    return response

@app.get("/metrics")
async def metrics():
    """Return Prometheus metrics."""
    return generate_latest()

---

Performance Optimization Checklist

# ✅ Async all the way
@app.post("/api/data")
async def process(data: MyModel):  # Not def, use async def
    result = await async_operation()
    return result

# ✅ Connection pooling
# Don't create new DB connections per request
# Use connection pools (20-50 connections)

# ✅ Caching hot data
# Store latest vehicle state in Redis
# TTL: 30 seconds for real-time data

# ✅ Compression
# Enable gzip compression on responses
app = FastAPI()
from fastapi.middleware.gzip import GZIPMiddleware
app.add_middleware(GZIPMiddleware, minimum_size=1000)

# ✅ Rate limiting
from slowapi import Limiter
limiter = Limiter(key_func=get_remote_address)
@app.post("/api/data")
@limiter.limit("1000/minute")
async def process(data: MyModel):
    pass

# ✅ Pagination for large datasets
@app.get("/api/analytics")
async def get_analytics(skip: int = 0, limit: int = 100):
    if limit > 1000:
        limit = 1000  # Cap maximum
    return await db.fetch(f"LIMIT {limit} OFFSET {skip}")

---

Deployment (Docker + Kubernetes)

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

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

# Copy app
COPY . .

# Run with Uvicorn (4 workers for 4 CPUs)
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# Kubernetes deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-ingestion
spec:
  replicas: 3
  selector:
    matchLabels:
      app: api-ingestion
  template:
    metadata:
      labels:
        app: api-ingestion
    spec:
      containers:
      - name: api
        image: vehicle-metrics/api:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /api/health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /api/health
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: app-secrets
              key: database-url
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: api-ingestion-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: api-ingestion
  minReplicas: 3
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

---

Real-World Gotchas

1. Connection Pool Exhaustion

# ❌ Wrong: Creates new connection each time
@app.get("/api/data")
async def get_data():
    db = create_connection()  # New connection!
    result = await db.query(...)
    return result

# ✅ Right: Reuse from pool
@app.get("/api/data")
async def get_data(db = Depends(get_database)):  # From pool
    result = await db.query(...)
    return result

2. Async All the Way

# ❌ Wrong: Blocking call in async context
@app.get("/api/data")
async def get_data():
    result = requests.get("http://...")  # BLOCKS!
    return result

# ✅ Right: Use async HTTP client
import httpx

@app.get("/api/data")
async def get_data():
    async with httpx.AsyncClient() as client:
        result = await client.get("http://...")
    return result

3. Memory Leaks with Background Tasks

# ❌ Wrong: Task references prevent garbage collection
background_tasks.add_task(
    process_data,
    large_data_object,  # Stays in memory!
)

# ✅ Right: Pass only what you need
background_tasks.add_task(
    process_data,
    large_data_object.id,  # Just the ID
)

---

TL;DR

• Choose FastAPI for real-time, high-throughput APIs
• Use async/await everywhere (not one sync function!)
• Validate with Pydantic (free type safety)
• Dependency injection keeps code clean and testable
• Connection pooling is non-negotiable
• Fire-and-forget for non-blocking work
• Logging + Metrics = visibility into production
• Test everything (TestClient makes it easy)
• Deploy on Kubernetes with auto-scaling

---

GitHub: beltagyy/vehicle-metrics
Author: Mohamed ElBeltagy (@beltagyy)
Topic: FastAPI | Python | Microservices