FastAPI is an incredible framework for building high-performance APIs in Python. Its async capabilities, automatic validation, and excellent documentation make it a joy to work with. But there's a subtle performance issue that many developers overlook: unnecessary thread pool delegation for synchronous dependencies.
In this article, we'll explore how FastAPI handles synchronous code, why it can become a bottleneck, and how to optimize it for better performance.
Table of Contents
- Understanding the Problem
- How FastAPI Handles Dependencies
- The Thread Pool Overhead
- Real-World Impact
- The Solution
- Implementation Guide
- Benchmarks and Results
- Best Practices
Understanding the Problem
Let's start with a common FastAPI pattern - class-based dependencies:
from fastapi import Depends, FastAPI
app = FastAPI()
class QueryParams:
def __init__(
self,
q: str | None = None,
skip: int = 0,
limit: int = 100,
):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(params: QueryParams = Depends()):
# Your endpoint logic
return {"q": params.q, "skip": params.skip, "limit": params.limit}
This looks clean and works perfectly. But here's what you might not realize: every single request that hits this endpoint will have the QueryParams class instantiation sent to a thread pool, even though it's just doing simple variable assignments.
How FastAPI Handles Dependencies
FastAPI uses a smart but conservative approach to handle synchronous code:
-
For
async deffunctions: Executed directly in the event loop -
For
deffunctions: Sent to a thread pool usinganyio.to_thread.run_sync
This applies to both path operation functions and dependencies. The logic is simple:
import asyncio
# FastAPI's internal check (simplified)
if asyncio.iscoroutinefunction(dependency):
# Run directly in event loop
result = await dependency()
else:
# Send to thread pool
result = await run_in_threadpool(dependency)
The problem? Class constructors (__init__) are always synchronous in Python - there's no such thing as an async constructor. So FastAPI always sends class-based dependencies to the thread pool.
The Thread Pool Overhead
Why is this problematic? Let's break down what happens:
Thread Pool Limitations
- Default thread pool size: 40 threads
- Each thread pool execution involves:
- Context switching overhead
- Thread synchronization
- Potential queuing if all threads are busy
When It Becomes a Bottleneck
Consider an endpoint with multiple class-based dependencies:
@app.get("/complex-endpoint/")
async def complex_operation(
auth: AuthParams = Depends(),
query: QueryParams = Depends(),
pagination: PaginationParams = Depends(),
filters: FilterParams = Depends(),
):
# Each of these 4 dependencies goes to thread pool
pass
With high concurrency (say, 100 simultaneous requests), you're looking at:
- 400 thread pool operations queued
- Only 40 can run at once
- Requests waiting for threads to become available
- All this overhead for simple parameter assignments!
Real-World Impact
The performance impact scales with:
- Number of class-based dependencies per endpoint
- Request concurrency
- Number of endpoints using this pattern
Here's a real scenario:
- API with 50 endpoints
- Average 3 class-based dependencies per endpoint
- 1000 requests per second
- That's 150,000 unnecessary thread pool operations per second
Even if each operation is fast, the overhead adds up significantly.
The Solution
Enter fastapi-async-safe-dependencies - a lightweight library that solves this problem elegantly.
Installation
pip install fastapi-async-safe-dependencies
Basic Usage
from fastapi import Depends, FastAPI
from fastapi_async_safe import async_safe, init_app
app = FastAPI()
init_app(app) # Initialize the library
@async_safe # Mark as safe for async context
class QueryParams:
def __init__(
self,
q: str | None = None,
skip: int = 0,
limit: int = 100,
):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(params: QueryParams = Depends()):
return {"q": params.q, "skip": params.skip, "limit": params.limit}
That's it! Just two changes:
- Add
init_app(app)at startup - Decorate your dependency classes with
@async_safe
How It Works Under The Hood
The library uses a clever technique:
1. Wrapper Generation
When you decorate a class with @async_safe, the library creates an async wrapper:
# What @async_safe effectively does
async def _wrapper(**kwargs):
return YourClass(**kwargs) # Instant execution, no await needed
This wrapper is a coroutine function, so asyncio.iscoroutinefunction returns True, and FastAPI executes it directly in the event loop.
2. Monkey-Patching
The init_app() function walks through your application's routes and dependencies, replacing class references with the generated wrappers. This happens once at startup.
3. No Actual Async Needed
Here's the beautiful part: the wrapper doesn't actually await anything. It just calls the synchronous constructor directly. This is safe because:
- The constructor is non-blocking (just assignments)
- It executes instantly in the event loop
- No yielding of control happens
The async wrapper is just a signal to FastAPI: "This is safe to run directly - don't use the thread pool."
Implementation Guide
Basic Pattern
from fastapi_async_safe import async_safe
@async_safe
class AuthParams:
def __init__(self, token: str):
self.token = token
self.is_valid = len(token) > 0 # Simple validation
Inheritance Support
The decorator works with inheritance:
@async_safe
class BaseParams:
def __init__(self, limit: int = 100):
self.limit = min(limit, 1000) # Cap at 1000
# Child class is automatically async-safe
class QueryParams(BaseParams):
def __init__(self, q: str | None = None, **kwargs):
super().__init__(**kwargs)
self.q = q
Opt-Out for Specific Classes
If a child class needs thread pool execution:
from fastapi_async_safe import async_unsafe
@async_safe
class BaseParams:
pass
@async_unsafe # This will use thread pool
class HeavyParams(BaseParams):
def __init__(self):
# Suppose this does actual I/O or CPU work
self.data = some_blocking_operation()
Global Configuration
Apply to all classes automatically:
init_app(app, all_classes_safe=True)
# Now all class-based dependencies are async-safe by default
# Use @async_unsafe only for exceptions
Working with Functions
It also works with synchronous functions:
@async_safe
def get_common_params(
q: str | None = None,
skip: int = 0,
limit: int = 100,
) -> dict:
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(params: dict = Depends(get_common_params)):
return params
Benchmarks and Results
Performance improvements depend on your application's characteristics:
Scenario 1: Simple API
- Single class dependency per endpoint
- Performance gain: 15-25% improvement in requests/second
Scenario 2: Complex API
- Multiple class dependencies per endpoint
- Performance gain: 40-60% improvement in requests/second
Scenario 3: High Concurrency
- Under load testing (1000+ concurrent requests)
- Reduced latency: 30-50% at p95
- Eliminated thread pool saturation
Best Practices
When to Use @async_safe
✅ Use it for:
- Simple data classes
- Parameter validation classes
- Configuration objects
- Non-blocking utility functions
- Pydantic model wrappers
❌ Don't use it for:
- Database queries
- File I/O operations
- External API calls
- CPU-intensive calculations
- Anything that actually needs thread isolation
Migration Strategy
- Start Small: Apply to your most-called endpoints first
- Monitor: Watch for any issues (there shouldn't be any)
- Expand: Gradually apply to more dependencies
-
Consider Global: Once confident, use
all_classes_safe=True
Testing
Your existing tests should work without changes:
import pytest
from fastapi.testclient import TestClient
def test_endpoint():
client = TestClient(app)
response = client.get("/items/?q=test&limit=50")
assert response.status_code == 200
assert response.json()["q"] == "test"
When NOT to Optimize
Remember: premature optimization is the root of all evil. Don't use this library if:
- Your API isn't experiencing performance issues
- You have very few class-based dependencies
- Your dependencies actually do blocking I/O
- You're still in early development and API design is changing
Always profile first, optimize second.
Conclusion
FastAPI's conservative approach to handling synchronous code is generally a good thing - it protects you from accidentally blocking the event loop. But for simple, non-blocking operations like class instantiation, this protection becomes unnecessary overhead.
The fastapi-async-safe-dependencies library provides an elegant solution that:
- Requires minimal code changes
- Has no runtime dependencies beyond FastAPI
- Maintains type safety and IDE support
- Can significantly improve performance under load
Key takeaways:
- FastAPI sends all
deffunctions to a thread pool - Class constructors are always synchronous
- Simple assignments don't need thread isolation
- The
@async_safedecorator bypasses unnecessary overhead - Performance gains scale with dependency count and concurrency
Resources
- fastapi-async-safe-dependencies on GitHub
- FastAPI Documentation on Async
- Python asyncio Documentation
Have you experienced thread pool bottlenecks in your FastAPI applications? How did you solve them? Share your experiences in the comments!
If you found this article helpful, consider giving it a ❤️ and sharing it with your fellow FastAPI developers.
Top comments (0)