From FastAPI to Microservices: Handling 10K Concurrent Requests

1┌────────────────────────────────────────┐
2│            Flask Monolith              │
3│  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐  │
4│  │ Auth │ │ User │ │ Task │ │ Data │  │
5│  └──────┘ └──────┘ └──────┘ └──────┘  │
6└────────────────────────────────────────┘

1┌─────────────┐
2│ API Gateway │
3└──────┬──────┘
4       │
5┌──────┴──────┬──────────────┬──────────────┐
6│             │              │              │
7▼             ▼              ▼              ▼
8┌─────┐   ┌──────┐    ┌──────┐    ┌──────┐
9│Auth │   │ User │    │ Task │    │ Data │
10│ API │   │ API  │    │ API  │    │ API  │
11└─────┘   └──────┘    └──────┘    └──────┘

1from fastapi import FastAPI
2from httpx import AsyncClient
3from sqlalchemy.ext.asyncio import AsyncSession
4
5app = FastAPI()
6
7# Bad: Blocking database call
8@app.get("/users/{user_id}")
9def get_user(user_id: int, db: Session = Depends(get_db)):
10    return db.query(User).filter(User.id == user_id).first()
11
12# Good: Async database call
13@app.get("/users/{user_id}")
14async def get_user(user_id: int, db: AsyncSession = Depends(get_async_db)):
15    result = await db.execute(select(User).where(User.id == user_id))
16    return result.scalar_one_or_none()

1from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
2from sqlalchemy.orm import sessionmaker
3
4engine = create_async_engine(
5    DATABASE_URL,
6    pool_size=20,
7    max_overflow=30,
8    pool_timeout=30,
9    pool_recycle=1800,
10)
11
12AsyncSessionLocal = sessionmaker(
13    engine,
14    class_=AsyncSession,
15    expire_on_commit=False
16)

1from fastapi_cache import FastAPICache
2from fastapi_cache.backends.redis import RedisBackend
3from fastapi_cache.decorator import cache
4
5@app.on_event("startup")
6async def startup():
7    redis = aioredis.from_url("redis://localhost")
8    FastAPICache.init(RedisBackend(redis), prefix="api-cache")
9
10@app.get("/products/{product_id}")
11@cache(expire=300)  # Cache for 5 minutes
12async def get_product(product_id: int):
13    # This result will be cached
14    return await fetch_product(product_id)

1# locustfile.py
2from locust import HttpUser, task, between
3
4class APIUser(HttpUser):
5    wait_time = between(0.1, 0.5)
6
7    @task(3)
8    def get_tasks(self):
9        self.client.get("/api/v1/tasks")
10
11    @task(1)
12    def create_task(self):
13        self.client.post("/api/v1/tasks", json={
14            "title": "Test task",
15            "priority": "high"
16        })

1from fastapi import HTTPException
2from pydantic import BaseModel
3
4class ErrorResponse(BaseModel):
5    error_code: str
6    message: str
7    details: dict | None = None
8
9@app.exception_handler(HTTPException)
10async def http_exception_handler(request, exc):
11    return JSONResponse(
12        status_code=exc.status_code,
13        content=ErrorResponse(
14            error_code=f"ERR_{exc.status_code}",
15            message=exc.detail,
16        ).dict()
17    )

1from circuitbreaker import circuit
2
3@circuit(failure_threshold=5, recovery_timeout=30)
4async def call_external_service(data: dict):
5    async with httpx.AsyncClient() as client:
6        response = await client.post(EXTERNAL_URL, json=data)
7        response.raise_for_status()
8        return response.json()

1@app.get("/health")
2async def health_check():
3    checks = {
4        "database": await check_database(),
5        "redis": await check_redis(),
6        "external_api": await check_external_api(),
7    }
8
9    status = "healthy" if all(checks.values()) else "degraded"
10    return {"status": status, "checks": checks}

1@app.get("/recommendations/{user_id}")
2async def get_recommendations(user_id: int):
3    try:
4        # Try personalized recommendations
5        return await ml_service.get_personalized(user_id)
6    except ServiceUnavailable:
7        # Fall back to popular items
8        return await get_popular_items()
9    except Exception:
10        # Ultimate fallback
11        return {"recommendations": [], "fallback": True}

1import structlog
2
3logger = structlog.get_logger()
4
5@app.middleware("http")
6async def logging_middleware(request: Request, call_next):
7    request_id = str(uuid.uuid4())
8
9    with structlog.contextvars.bound_contextvars(
10        request_id=request_id,
11        path=request.url.path,
12        method=request.method,
13    ):
14        logger.info("request_started")
15
16        start = time.perf_counter()
17        response = await call_next(request)
18        duration = time.perf_counter() - start
19
20        logger.info(
21            "request_completed",
22            status_code=response.status_code,
23            duration_ms=round(duration * 1000, 2)
24        )
25
26        return response

1from prometheus_fastapi_instrumentator import Instrumentator
2
3Instrumentator().instrument(app).expose(app)