Race Condition in Fastapi

Race conditions in FastAPI applications typically emerge from concurrent requests manipulating shared state without proper synchronization. FastAPI's async nature and high performance make it particularly susceptible to these issues when developers assume sequential execution.

A classic FastAPI race condition occurs in inventory management endpoints. Consider a shopping cart implementation where two concurrent requests attempt to purchase the last item in stock:

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Dict

app = FastAPI()

inventory: Dict[int, int] = {1: 1}  # Item 1 has 1 unit in stock

class PurchaseRequest(BaseModel):
    item_id: int
    quantity: int

@app.post("/purchase")
async def purchase(request: PurchaseRequest):
    if inventory[request.item_id] >= request.quantity:
        inventory[request.item_id] -= request.quantity
        return {"status": "success"}
    return {"status": "out of stock"}

Two simultaneous requests checking inventory[1] >= 1 both see 1 available, both pass the condition, and both decrement to -1. FastAPI's async event loop allows this interleaving because the inventory check and update aren't atomic.

FastAPI's dependency injection system can mask race conditions. When using Depends() with shared resources, multiple requests may resolve dependencies concurrently:

from fastapi import Depends

def get_user_session():
    # Shared session object
    return user_session

@app.post("/transfer")
async def transfer_funds(
    request: TransferRequest,
    session: Session = Depends(get_user_session)
):
    balance = session.get_balance(request.from_account)
    if balance >= request.amount:
        session.update_balance(request.from_account, balance - request.amount)
        session.update_balance(request.to_account, request.amount)
        return {"status": "transferred"}

If get_user_session() returns a shared database session, concurrent transfers can corrupt balances. FastAPI's default behavior is to create a new event loop for each request, but shared resources like database connections or in-memory state remain vulnerable.

Database-level race conditions are particularly dangerous in FastAPI. Using async database drivers without proper transaction isolation:

from databases import Database
import asyncio

@app.post("/update_profile")
async def update_profile(request: ProfileUpdate):
    # Two concurrent requests
    await database.execute(
        "UPDATE users SET bio = $1 WHERE id = $2", 
        request.bio, request.user_id
    )
    # Another request updates email
    await database.execute(
        "UPDATE users SET email = $1 WHERE id = $2",
        request.email, request.user_id
    )

Without transaction boundaries, these updates can interleave, causing partial profile updates or lost changes.

Detecting race conditions in FastAPI requires understanding the async execution model and identifying shared mutable state. middleBrick's scanning engine specifically targets FastAPI's async patterns and dependency injection system.

middleBrick identifies race condition vulnerabilities by analyzing FastAPI route handlers for:

• Shared mutable state accessed without locks (dictionaries, lists, counters)
• Database operations without transaction boundaries
• Async endpoints that read-modify-write without atomic operations
• Shared resource dependencies injected via Depends()

The scanner examines your FastAPI application's runtime behavior, not just static code. It submits concurrent requests to the same endpoint and monitors for inconsistent responses or state corruption.

For FastAPI applications, middleBrick specifically checks:

# Race condition detection patterns
# Shared state without synchronization
@app.post("/increment")
async def increment_counter():
    counter["count"] += 1  # Vulnerable: no lock

# Database race conditions
@app.post("/transfer")
async def transfer_funds():
    # No transaction - vulnerable to interleaving
    await db.execute("UPDATE accounts SET balance = balance - $1 WHERE id = $2", amount, from_id)
    await db.execute("UPDATE accounts SET balance = balance + $1 WHERE id = $2", amount, to_id)

middleBrick's LLM security module also detects AI-specific race conditions in FastAPI applications using large language models. When multiple requests simultaneously interact with LLM endpoints, prompt injection attacks can exploit timing gaps:

# Vulnerable AI endpoint
@app.post("/chat")
async def chat_with_ai(prompt: str):
    response = await llm_client.generate(prompt)
    return {"response": response}

Concurrent requests might manipulate the LLM's context window or cause prompt injection through timing-based attacks.

middleBrick's OpenAPI analysis identifies FastAPI-specific patterns that commonly lead to race conditions, such as:

• Endpoints using Depends() with shared database sessions
• Async routes with state-modifying operations
• Endpoints lacking proper validation before state changes

The scanner provides FastAPI-specific remediation guidance, including recommendations for using database transactions, async locks, or atomic operations that align with FastAPI's async/await patterns.

Remediating race conditions in FastAPI requires leveraging async-specific synchronization primitives and understanding FastAPI's async execution model. The most effective approach combines database-level transactions with application-level locking where appropriate.

For inventory management race conditions, use database transactions with proper isolation levels:

from fastapi import FastAPI, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.future import select
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

@app.post("/purchase")
async def purchase(request: PurchaseRequest):
    async with AsyncSessionLocal() as session:
        async with session.begin():
            # SELECT ... FOR UPDATE locks the row
            result = await session.execute(
                select(Inventory).where(Inventory.item_id == request.item_id).with_for_update()
            )
            inventory = result.scalar_one_or_none()
            
            if inventory and inventory.quantity >= request.quantity:
                inventory.quantity -= request.quantity
                await session.commit()
                return {"status": "success"}
            raise HTTPException(status_code=400, detail="Insufficient stock")

The with_for_update() clause locks the inventory row, preventing concurrent modifications. FastAPI's async context manager ensures proper cleanup.

For in-memory state race conditions, use asyncio locks:

import asyncio
from fastapi import FastAPI

app = FastAPI()
counter = 0
counter_lock = asyncio.Lock()

@app.post("/safe_increment")
async def safe_increment():
    async with counter_lock:
        global counter
        counter += 1
        return {"count": counter}

The async with counter_lock ensures only one coroutine modifies the counter at a time, preventing race conditions.

For FastAPI dependency injection race conditions, use scoped dependencies:

from fastapi import Depends, FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def get_database_session():
    session = AsyncSessionLocal()
    try:
        yield session
        await session.commit()
    finally:
        await session.close()

@app.post("/transfer")
async def transfer_funds(
    request: TransferRequest,
    session: AsyncSession = Depends(get_database_session)
):
    # Each request gets its own session - no shared state
    async with session.begin():
        # Proper transaction handling
        pass

This pattern ensures each request gets a fresh database session, eliminating shared state race conditions.

For AI-specific race conditions in FastAPI LLM endpoints, implement request queuing and rate limiting:

from fastapi import FastAPI, BackgroundTasks
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

app = FastAPI()
limiter = Limiter(key_func=get_remote_address)
app.add_exception_handler(
    _rate_limit_exceeded_handler
)

@app.post("/chat")
@limiter.limit("5/minute")
async def chat_with_ai(prompt: str):
    # Queue requests to prevent prompt injection race conditions
    async with request_queue:
        response = await llm_client.generate(prompt)
    return {"response": response}

Rate limiting prevents concurrent requests from overwhelming the LLM endpoint, reducing the window for race condition exploitation.