Introduction
Welcome to Module 13 of the FastAPI Complete Course. In this chapter, we dive deep into FastAPI Async Programming. If you have been following along, you know FastAPI is built on top of Starlette and leverages Python’s asyncio to handle thousands of concurrent connections with ease. Understanding async programming is not just a “nice to have” for FastAPI developers—it is essential for building high-performance, production-ready APIs.
In this tutorial, we will start from the very basics: the difference between synchronous and asynchronous code. Then we will progress through writing async functions, performing async database operations, and understanding the performance benefits. By the end, you will have a solid grasp of concurrency concepts and be able to write efficient, non-blocking FastAPI applications. Let’s begin.
Sync vs Async: The Core Difference
What is Synchronous Code?
Synchronous (or sync) code executes one task at a time. When a function makes a request to a database or an external API, the entire program waits for that operation to complete before moving to the next line. This is often called “blocking” code.
Consider a simple example: you order coffee at a busy shop. You stand in line, place your order, wait for the barista to make it, and only then you step aside. While you wait, you are blocking the line behind you. This is synchronous behavior.
What is Asynchronous Code?
Asynchronous (async) code allows a program to start a long-running operation and continue executing other tasks while waiting for it to finish. In the coffee shop analogy, you place your order, get a buzzer, and sit down. While the barista makes your coffee, you can read a book, check your phone, or chat. When your buzzer rings, you pick up your coffee. The line is never blocked, and the barista works on multiple orders simultaneously.
In Python, asyncio is the library that enables this behavior. FastAPI is built on top of it, meaning every endpoint you write can be async by default.
Code Example: Sync vs Async in FastAPI
Let’s see a practical comparison. Below are two FastAPI endpoints—one sync, one async—that simulate a 2-second delay (like a slow database query).
# sync_example.py
import time
from fastapi import FastAPI
app = FastAPI()
@app.get("/sync")
def read_sync():
time.sleep(2) # Simulates a blocking I/O operation
return {"message": "Synchronous endpoint completed"}
@app.get("/async")
async def read_async():
await asyncio.sleep(2) # Non-blocking sleep
return {"message": "Asynchronous endpoint completed"}Explanation:
time.sleep(2)blocks the entire thread. If 10 users hit the sync endpoint simultaneously, they will be processed one after another. The last user waits 20 seconds.await asyncio.sleep(2)pauses only the current coroutine. While one user’s request is “sleeping,” FastAPI can handle other requests. All 10 users get a response in roughly 2 seconds total.
Async Functions in FastAPI
Defining an Async Route
FastAPI makes it incredibly easy to define async routes. Simply add the async keyword before def, and use await inside the function for any I/O operations.
from fastapi import FastAPI
import httpx # For async HTTP requests
app = FastAPI()
@app.get("/weather/{city}")
async def get_weather(city: str):
# Simulate fetching weather from an external API
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.weather.com/v1/{city}")
data = response.json()
return {"city": city, "temperature": data["temp"]}Line-by-line explanation:
async def get_weather(city: str):— This declares the endpoint as asynchronous. FastAPI will run it in an event loop.async with httpx.AsyncClient() as client:— We create an async HTTP client. Theasync withensures proper resource cleanup.response = await client.get(...)— Theawaitkeyword yields control back to the event loop while the HTTP request is in flight. The server can handle other requests during this time.
Using await Correctly
You can only use await inside an async def function. Attempting to use it in a regular def will raise a SyntaxError. Also, you cannot call an async function directly without await—it returns a coroutine object, not the result.
# Wrong: calling async function without await
result = get_weather("London") # Returns a coroutine, not data
# Correct
result = await get_weather("London")Async Service Class Example
For larger applications, it is best practice to encapsulate business logic in service classes. Here is an async service class that handles user creation:
# services/user_service.py
import asyncio
from typing import Dict
class UserService:
def __init__(self):
self.users_db = {} # Simulated in-memory database
async def create_user(self, username: str, email: str) -> Dict:
# Simulate async I/O (e.g., writing to a database)
await asyncio.sleep(0.1) # Simulates a non-blocking write
user_id = len(self.users_db) + 1
user = {"id": user_id, "username": username, "email": email}
self.users_db[user_id] = user
return user
async def get_user(self, user_id: int) -> Dict:
await asyncio.sleep(0.05) # Simulates a non-blocking read
return self.users_db.get(user_id, {})Now use this service in a FastAPI route:
# main.py
from fastapi import FastAPI
from services.user_service import UserService
app = FastAPI()
user_service = UserService()
@app.post("/users/")
async def create_user(username: str, email: str):
user = await user_service.create_user(username, email)
return user
@app.get("/users/{user_id}")
async def get_user(user_id: int):
user = await user_service.get_user(user_id)
if not user:
return {"error": "User not found"}
return userWhy use a service class? It separates concerns, makes testing easier, and allows you to swap out the simulated database for a real one without changing your route logic.
Async Database Operations
Why Async Database Drivers Matter
Traditional database drivers like psycopg2 (for PostgreSQL) are synchronous. When you call cursor.execute(), the thread blocks until the database responds. In a high-concurrency API, this blocks the event loop and kills performance.
Async database drivers (e.g., asyncpg for PostgreSQL, aiomysql for MySQL, motor for MongoDB) are specifically designed to work with asyncio. They release the event loop while waiting for the database, allowing other requests to be processed.
Example: Async PostgreSQL with asyncpg
First, install the driver:
pip install asyncpgNow, create a database connection pool and use it in a FastAPI endpoint:
# database.py
import asyncpg
from fastapi import FastAPI
app = FastAPI()
# Create a connection pool at startup
@app.on_event("startup")
async def startup():
app.state.pool = await asyncpg.create_pool(
user="postgres",
password="password",
database="fastapi_db",
host="localhost",
port=5432,
min_size=5,
max_size=20
)
@app.on_event("shutdown")
async def shutdown():
await app.state.pool.close()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
async with app.state.pool.acquire() as connection:
# The query is non-blocking
row = await connection.fetchrow(
"SELECT id, name, price FROM items WHERE id = $1", item_id
)
if row:
return {"id": row["id"], "name": row["name"], "price": row["price"]}
return {"error": "Item not found"}Key points:
async with app.state.pool.acquire() as connection:— Acquires a connection from the pool asynchronously. If all connections are busy, it waits without blocking the event loop.await connection.fetchrow(...)— Sends the query and awaits the result. The event loop can handle other requests during this wait.$1— Parameterized query syntax for asyncpg to prevent SQL injection.
Async ORMs: SQLAlchemy Async
If you prefer an ORM, SQLAlchemy 1.4+ supports async sessions. Here is a minimal example:
# models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
price = Column(Integer)# main.py
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from models import Item
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
app = FastAPI()
async def get_session() -> AsyncSession:
async with async_session() as session:
yield session
@app.get("/items/{item_id}")
async def read_item(item_id: int, session: AsyncSession = Depends(get_session)):
result = await session.get(Item, item_id)
if result:
return {"id": result.id, "name": result.name, "price": result.price}
return {"error": "Not found"}Note: The get_session dependency uses async with to ensure the session is properly closed after the request.
Performance Benefits of Async Programming
Throughput vs. Latency
Async programming primarily improves throughput—the number of requests your server can handle per second. It does not make a single request faster (latency remains similar), but it allows the server to handle many requests concurrently without creating a new thread or process for each.
Consider a scenario where each request involves a 100ms database query. With a synchronous server (like Flask’s default), one worker can handle about 10 requests per second. With an async server (like FastAPI with Uvicorn), a single worker can handle hundreds or even thousands of requests per second because it does not sit idle waiting for the database.
Real-World Impact
In production, this means you can serve the same traffic with fewer server resources (CPU and memory). This translates to lower cloud costs and better user experience under load.
Benchmarking Example
You can test this yourself using a tool like wrk or locust. Here is a simple benchmark using ab (Apache Bench) on a local machine:
# Test sync endpoint (assuming you have the sync example running)
ab -n 100 -c 10 http://127.0.0.1:8000/sync
# Test async endpoint
ab -n 100 -c 10 http://127.0.0.1:8000/asyncYou will likely see that the async endpoint completes all 100 requests significantly faster, especially with higher concurrency (-c flag).
Concurrency Concepts
Event Loop
The event loop is the core of async programming. It is a loop that continuously checks for tasks that are ready to run. When you await something, the event loop pauses that coroutine and runs another one that is ready. Once the awaited operation completes, the event loop resumes the original coroutine.
In FastAPI, Uvicorn runs the event loop. You rarely need to interact with it directly, but understanding it helps you debug performance issues.
Coroutines
A coroutine is a special function declared with async def. When called, it returns a coroutine object. To execute it, you must either await it or schedule it with asyncio.create_task().
async def fetch_data():
await asyncio.sleep(1)
return {"data": "important"}
# This does NOT run the function
coro = fetch_data() # Returns a coroutine object
# This runs it
result = await fetch_data()Tasks vs. Coroutines
A task is a coroutine wrapped in a asyncio.Task. Tasks are scheduled to run concurrently on the event loop. Use asyncio.create_task() to run multiple coroutines “in the background.”
async def background_work():
await asyncio.sleep(5)
print("Background task completed")
@app.get("/start-task")
async def start_task():
# This task runs concurrently, not blocking the response
asyncio.create_task(background_work())
return {"message": "Task started, response sent immediately"}Caution: If the server shuts down before the task finishes, it will be cancelled. For critical background tasks, use a task queue like Celery or ARQ.
Concurrency vs. Parallelism
It is crucial to understand that async concurrency is not parallelism. Python’s Global Interpreter Lock (GIL) prevents true parallel execution of Python code across multiple CPU cores. Async programming achieves concurrency by interleaving tasks on a single thread. This is perfect for I/O-bound tasks (network requests, database queries, file reads). For CPU-bound tasks (image processing, heavy calculations), you still need multiprocessing.
Common Mistakes
- Blocking the event loop with sync calls: Accidentally using
time.sleep()or a sync database driver inside an async route. This blocks the entire server. Always use async alternatives. - Forgetting
await: Calling an async function withoutawaitreturns a coroutine object, not the result. This often leads to confusing errors like “coroutine object is not callable“. - Using
awaitin a non-async function: You cannot useawaitinside a regulardef. If you need to call an async function from a sync context, useasyncio.run()(but be careful—this creates a new event loop). - Not using connection pools: Opening a new database connection for every request is slow and resource-intensive. Always use a pool (e.g.,
asyncpg.create_pool). - Ignoring background task lifecycle: Tasks created with
asyncio.create_task()are not tracked. If the server restarts, they are lost. For critical tasks, use a persistent task queue.
Practice Task
Now it is your turn to apply what you have learned. Build a small FastAPI application with the following requirements:
- Create an async service class called
ProductServicethat simulates a database with an in-memory dictionary. - The service should have async methods:
create_product(name, price),get_product(product_id), andlist_products(). - Create three FastAPI endpoints that use this service:
POST /products/,GET /products/{product_id}, andGET /products/. - Add a background task that logs “Product created” 3 seconds after a product is created. Use
asyncio.create_task(). - Test your endpoints using
curlor the interactive docs at/docs.
Bonus: If you have a PostgreSQL database available, replace the in-memory dictionary with asyncpg and a real database.
FAQs
1. Do I need to make every FastAPI route async?
No. If a route does not perform any I/O (e.g., it just returns a static dictionary), a sync route is fine. However, making it async does not hurt. FastAPI automatically runs sync routes in a thread pool so they do not block the event loop.
2. Can I use async with Django or Flask?
Django 3.1+ has limited async support, but it is not as seamless as FastAPI. Flask does not natively support async routes (though there are extensions). FastAPI is built from the ground up for async.
3. What is the difference between await and asyncio.run()?
await is used inside an async function to call another async function. asyncio.run() is a top-level function that creates a new event loop, runs a single coroutine, and closes the loop. You typically use asyncio.run() in scripts, not inside FastAPI endpoints.
4. Will async make my CPU-bound code faster?
No. Async is for I/O-bound tasks. For CPU-bound tasks (e.g., image processing, data analysis), use multiprocessing (e.g., concurrent.futures.ProcessPoolExecutor).
5. How do I debug async code?
Use print() statements or logging. For more advanced debugging, use asyncio.run(debug=True) or an IDE debugger that supports async (PyCharm, VS Code).
Summary
In this module, we covered the essentials of FastAPI Async Programming. You learned:
- The fundamental difference between sync and async code.
- How to write async routes and use
awaitcorrectly. - How to build async service classes for clean architecture.
- How to perform async database operations using asyncpg and SQLAlchemy async.
- The performance benefits of async: higher throughput with fewer resources.
- Key concurrency concepts: event loop, coroutines, tasks, and the difference between concurrency and parallelism.
Async programming is a superpower in modern web development. With FastAPI, you can build APIs that are fast, scalable, and maintainable. Practice the exercises, experiment with different database drivers, and you will be ready for production-level applications.
In Module 14: WebSockets and Real-Time Features, we will take your async skills to the next level by building real-time chat applications and live data streams. See you there!
More Practical Examples
Let’s dive deeper into real-world async patterns you’ll use daily in FastAPI. The following examples demonstrate common scenarios where async truly shines.
Example 1: Parallel API Calls
Imagine you need to fetch data from three different external APIs to build a dashboard. Without async, each request would wait for the previous one to complete. With async, they run concurrently.
import httpx
from fastapi import FastAPI
app = FastAPI()
# Simulate external API endpoints
USERS_API = "https://jsonplaceholder.typicode.com/users"
POSTS_API = "https://jsonplaceholder.typicode.com/posts"
COMMENTS_API = "https://jsonplaceholder.typicode.com/comments"
async def fetch_users():
async with httpx.AsyncClient() as client:
response = await client.get(USERS_API)
return response.json()
async def fetch_posts():
async with httpx.AsyncClient() as client:
response = await client.get(POSTS_API)
return response.json()
async def fetch_comments():
async with httpx.AsyncClient() as client:
response = await client.get(COMMENTS_API)
return response.json()
@app.get("/dashboard")
async def get_dashboard():
# All three requests run concurrently
users, posts, comments = await asyncio.gather(
fetch_users(),
fetch_posts(),
fetch_comments()
)
return {
"users_count": len(users),
"posts_count": len(posts),
"comments_count": len(comments)
}Explanation: The asyncio.gather() function runs all three fetch functions concurrently. Each function creates its own AsyncClient session, but the key insight is that while one request is waiting for the network response, Python can switch to another request. This reduces total execution time from the sum of all three requests to approximately the time of the slowest one.
Example 2: Background Tasks with Async
Sometimes you need to start a long-running process and return a response immediately. FastAPI’s BackgroundTasks works perfectly with async functions.
from fastapi import FastAPI, BackgroundTasks
import asyncio
import time
app = FastAPI()
async def send_welcome_email(email: str):
"""Simulate sending an email (takes 3 seconds)"""
await asyncio.sleep(3) # Simulate network delay
print(f"Welcome email sent to {email} at {time.strftime('%H:%M:%S')}")
async def log_signup(username: str):
"""Simulate logging to a database (takes 1 second)"""
await asyncio.sleep(1)
print(f"Signup logged for {username} at {time.strftime('%H:%M:%S')}")
@app.post("/signup")
async def signup(username: str, email: str, background_tasks: BackgroundTasks):
# Add tasks to run in the background
background_tasks.add_task(send_welcome_email, email)
background_tasks.add_task(log_signup, username)
# Return immediately without waiting for background tasks
return {
"message": f"User {username} created successfully",
"note": "Welcome email will be sent shortly"
}Explanation: When a user signs up, the endpoint returns a response immediately. The background tasks (sending email and logging) run asynchronously after the response is sent. Notice how both tasks use await internally, but the caller doesn’t wait for them. This pattern is ideal for operations that don’t need to block the user’s response.
Class-Based Example
As your FastAPI application grows, you’ll want to organize async logic into classes. Here’s a practical example of an async service class that handles database operations.
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, select
from typing import List, Optional
# Database setup
DATABASE_URL = "sqlite+aiosqlite:///./blog.db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()
# Model
class Post(Base):
__tablename__ = "posts"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
content = Column(String)
author = Column(String)
# Async service class
class PostService:
def __init__(self, db: AsyncSession):
self.db = db
async def create_post(self, title: str, content: str, author: str) -> Post:
"""Create a new blog post"""
new_post = Post(title=title, content=content, author=author)
self.db.add(new_post)
await self.db.commit()
await self.db.refresh(new_post)
return new_post
async def get_post(self, post_id: int) -> Optional[Post]:
"""Get a single post by ID"""
result = await self.db.execute(select(Post).where(Post.id == post_id))
return result.scalar_one_or_none()
async def get_all_posts(self, skip: int = 0, limit: int = 10) -> List[Post]:
"""Get all posts with pagination"""
result = await self.db.execute(
select(Post).offset(skip).limit(limit)
)
return result.scalars().all()
async def update_post(self, post_id: int, title: str, content: str) -> Post:
"""Update an existing post"""
post = await self.get_post(post_id)
if not post:
raise HTTPException(status_code=404, detail="Post not found")
post.title = title
post.content = content
await self.db.commit()
await self.db.refresh(post)
return post
async def delete_post(self, post_id: int) -> bool:
"""Delete a post by ID"""
post = await self.get_post(post_id)
if not post:
raise HTTPException(status_code=404, detail="Post not found")
await self.db.delete(post)
await self.db.commit()
return True
# Dependency to get DB session
async def get_db():
async with AsyncSessionLocal() as session:
yield session
# FastAPI app
app = FastAPI()
@app.on_event("startup")
async def startup():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
@app.post("/posts/")
async def create_post(
title: str, content: str, author: str,
db: AsyncSession = Depends(get_db)
):
service = PostService(db)
post = await service.create_post(title, content, author)
return post
@app.get("/posts/")
async def read_posts(
skip: int = 0, limit: int = 10,
db: AsyncSession = Depends(get_db)
):
service = PostService(db)
posts = await service.get_all_posts(skip, limit)
return posts
@app.get("/posts/{post_id}")
async def read_post(post_id: int, db: AsyncSession = Depends(get_db)):
service = PostService(db)
post = await service.get_post(post_id)
if not post:
raise HTTPException(status_code=404, detail="Post not found")
return post
@app.put("/posts/{post_id}")
async def update_post(
post_id: int, title: str, content: str,
db: AsyncSession = Depends(get_db)
):
service = PostService(db)
updated_post = await service.update_post(post_id, title, content)
return updated_post
@app.delete("/posts/{post_id}")
async def delete_post(post_id: int, db: AsyncSession = Depends(get_db)):
service = PostService(db)
await service.delete_post(post_id)
return {"message": "Post deleted successfully"}Explanation: This class-based approach encapsulates all database operations for posts into a single PostService class. Each method is async and uses await for database operations. The class receives an async database session through dependency injection. This pattern makes your code more organized, testable, and reusable. Notice how error handling (like the 404 for missing posts) is handled within the service methods themselves.
To run this example, you need the async SQLite driver:
pip install aiosqlite sqlalchemy[asyncio]Step-by-Step Exercise
Let’s build a simple async web scraper that fetches multiple URLs and measures performance. This exercise will solidify your understanding of async programming.
Objective
Create a FastAPI endpoint that fetches content from multiple URLs concurrently and returns the response times.
Step 1: Setup
Create a new Python file called async_scraper.py and add the basic imports:
import asyncio
import time
from fastapi import FastAPI
import httpx
app = FastAPI()Step 2: Create the Async Fetch Function
Write an async function that fetches a single URL and measures the time:
async def fetch_url(url: str, client: httpx.AsyncClient) -> dict:
"""Fetch a URL and return its status and response time"""
start_time = time.time()
try:
response = await client.get(url, timeout=10.0)
elapsed = round(time.time() - start_time, 3)
return {
"url": url,
"status": response.status_code,
"response_time": elapsed,
"content_length": len(response.text)
}
except Exception as e:
elapsed = round(time.time() - start_time, 3)
return {
"url": url,
"status": "error",
"response_time": elapsed,
"error": str(e)
}Step 3: Create the Endpoint
Now create the endpoint that fetches multiple URLs:
@app.get("/scrape")
async def scrape_urls():
"""Fetch multiple URLs concurrently and return results"""
# List of URLs to scrape
urls = [
"https://jsonplaceholder.typicode.com/posts/1",
"https://jsonplaceholder.typicode.com/posts/2",
"https://jsonplaceholder.typicode.com/posts/3",
"https://jsonplaceholder.typicode.com/comments/1",
"https://jsonplaceholder.typicode.com/albums/1"
]
total_start = time.time()
# Use a single client session for efficiency
async with httpx.AsyncClient() as client:
# Create a list of tasks
tasks = [fetch_url(url, client) for url in urls]
# Run all tasks concurrently
results = await asyncio.gather(*tasks)
total_time = round(time.time() - total_start, 3)
return {
"total_time": total_time,
"urls_fetched": len(urls),
"results": results
}Step 4: Run and Test
Run your FastAPI app and test the endpoint:
uvicorn async_scraper:app --reloadOpen your browser and visit http://localhost:8000/scrape. You should see something like:
{
"total_time": 0.523,
"urls_fetched": 5,
"results": [
{"url": "https://jsonplaceholder.typicode.com/posts/1", "status": 200, "response_time": 0.234, "content_length": 292},
{"url": "https://jsonplaceholder.typicode.com/posts/2", "status": 200, "response_time": 0.245, "content_length": 292},
{"url": "https://jsonplaceholder.typicode.com/posts/3", "status": 200, "response_time": 0.198, "content_length": 292},
{"url": "https://jsonplaceholder.typicode.com/comments/1", "status": 200, "response_time": 0.312, "content_length": 449},
{"url": "https://jsonplaceholder.typicode.com/albums/1", "status": 200, "response_time": 0.289, "content_length": 98}
]
}Step 5: Challenge Yourself
Now try these extensions:
- Add a query parameter to accept custom URLs
- Implement a timeout per URL (hint: use
asyncio.wait_for()) - Add error handling for failed requests
- Compare performance with a synchronous version using
requestslibrary
Interview and Job Use Cases
Understanding async programming is crucial for FastAPI interviews and real-world jobs. Here’s what you need to know:
Common Interview Questions
Q: What’s the difference between concurrency and parallelism?
A: Concurrency is about dealing with many things at once (like a juggler keeping multiple balls in the air). Parallelism is about doing many things at once (like having multiple jugglers). Async programming in Python gives you concurrency, not true parallelism (unless you use multiprocessing).
Q: When should you NOT use async in FastAPI?
A: For CPU-bound tasks like image processing, video encoding, or complex calculations. These block the event loop. Use BackgroundTasks with a thread pool executor or a separate worker process instead.
Q: How do you handle database transactions in async FastAPI?
A: Use async database libraries like SQLAlchemy 1.4+ with async drivers (aiosqlite, asyncpg for PostgreSQL). Use async with context managers for transactions:
async with db.begin():
await db.execute(...)
await db.commit()Real-World Job Scenarios
Scenario 1: Real-time Dashboard
A company needs a dashboard that shows live data from 10 different microservices. Using async, you can fetch all endpoints concurrently, reducing load time from 10 seconds to ~1 second.
Scenario 2: File Upload Processing
Users upload CSV files that need validation, parsing, and database insertion. Use async to process multiple files concurrently, with background tasks for heavy operations.
Scenario 3: WebSocket Chat Application
Async is essential for WebSocket connections. Each connection is an async task that waits for messages without blocking other users.
Performance Optimization Tips for Interviews
- Connection Pooling: Reuse HTTP clients and database sessions to avoid connection overhead
- Bounded Semaphores: Limit concurrent operations to prevent overwhelming external services:
import asyncio
semaphore = asyncio.Semaphore(10) # Max 10 concurrent requests
async def safe_fetch(url):
async with semaphore:
async with httpx.AsyncClient() as client:
return await client.get(url)Extra Beginner FAQs
Q: Do I need to make ALL my functions async?
No! Only make functions async when they perform I/O operations (network requests, file reads, database queries). CPU-bound functions should remain synchronous or use thread pools. Mixing sync and async is fine – FastAPI handles this automatically.
Q: What happens if I call a sync function inside an async function?
It blocks the event loop. For example, calling time.sleep(5) inside an async endpoint would freeze your entire FastAPI server for 5 seconds. Always use await asyncio.sleep(5) instead. If you must use a sync library, wrap it with asyncio.to_thread():
import asyncio
import time
async def my_async_endpoint():
# This would block:
# time.sleep(5)
# This is correct:
await asyncio.to_thread(time.sleep, 5)
return {"message": "Done"}Q: Can I use async with databases other than SQLite?
Yes! Here are common async database drivers:
# PostgreSQL
pip install asyncpg
# MySQL
pip install aiomysql
# MongoDB
pip install motorQ: Why does my async code sometimes run slower than sync?
This usually happens when you:
- Have very few I/O operations (async overhead outweighs benefits)
- Use CPU-bound operations that block the event loop
- Don’t use
awaitproperly (forgetting it creates coroutines that never run)
Q: How do I debug async code?
Use these techniques:
- Enable asyncio debug mode:
PYTHONASYNCIODEBUG=1 uvicorn main:app - Add logging with timestamps to track execution order
- Use
asyncio.all_tasks()to see all running tasks - Set
loop.set_debug(True)for detailed coroutine information
Q: What’s the difference between async def and def in FastAPI?
FastAPI automatically handles both. If you use async def, FastAPI runs it in the event loop. If you use regular def, FastAPI runs it in a thread pool. For most I/O operations, async def is better because it doesn’t use extra threads. For CPU-bound work, regular def is actually better because it uses threads and doesn’t block the event loop.
Q: Can I use async with file operations?
Yes, but Python’s built-in file I/O isn’t truly async. Use aiofiles for async file operations:
pip install aiofilesimport aiofiles
async def read_file_async(filename: str):
async with aiofiles.open(filename, mode='r') as f:
contents = await f.read()
return contentsThis is useful when reading large files while serving other requests concurrently.
