The previous chapters built substantial functionality—full CRUD operations for posts and users, cascade deletions, PUT and PATCH endpoints with proper validation. Before adding more features, this chapter optimizes the existing foundation by converting the application from synchronous to asynchronous. The next chapter will reorganize code with routers to manage the growing main.py file, then forms and authentication will be added. This polish-then-build approach mirrors how real projects typically grow.
Understanding Asynchronous Programming
Async allows programs to handle multiple tasks concurrently. The classic analogy: synchronous code is like Subway—they make your entire sandwich from start to finish before moving to the next customer. Asynchronous code is like McDonald’s—someone takes your order and moves to the next customer while your food is being made in the background.
What async avoids: Waiting for external operations like database responses, network requests, or file reads. During that waiting time, other work can be done. These are called I/O-bound tasks—the situations where async improves performance:
- Database queries — waiting for the database to respond
- External API calls — waiting for network responses
- File operations — waiting for disk reads/writes
All these involve waiting, not computing. Async does not help with CPU-bound operations like heavy calculations, image processing, or data crunching. These keep the CPU busy doing actual work—there’s no waiting, so nothing for async to optimize.
Critical misconception: Async isn’t always faster. For simple, fast queries, you might not see much benefit. The overhead of async machinery can even slow things down slightly. The real benefits appear under concurrent load—many requests happening simultaneously.
How FastAPI Handles Sync vs Async
FastAPI’s handling of both approaches is nuanced and intelligent:
- Regular
defroutes: FastAPI automatically runs them in a separate thread pool. This prevents the function from blocking the main event loop. Even with regulardeffunctions, other requests can still be processed. This is automatic and works well. async defroutes: FastAPI runs them directly in the main event loop. This is more efficient, but you mustawaitany I/O operations. If you do blocking I/O withoutawait, you’ll block the entire event loop—this is worse than just using a regulardeffunction.
Both approaches work. Using regular def routes isn’t wrong or slow—FastAPI handles them intelligently. Choose based on what your specific route does. If you use async, ensure you’re using it correctly.
This is why the series started with synchronous routes. FastAPI does synchronous well, and the conversion happens now that the foundation is solid.
Installing Dependencies
For SQLite async support, install aiosqlite:
uv add aiosqliteaiosqlite provides an async driver for SQLite. SQLAlchemy can then use this driver for async operations. The same concept applies to other databases—for PostgreSQL, use asyncpg.
Additionally, install greenlet (required for SQLAlchemy’s async mode but not always automatically installed):
uv add greenletConverting Database Configuration
Update database.py to support async operations. Walk through each change to understand the differences.
Updated Imports
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBaseInstead of importing from sqlalchemy, import from sqlalchemy.ext.asyncio for AsyncSession, async_sessionmaker, and create_async_engine. DeclarativeBase still comes from sqlalchemy.orm.
Database URL with Async Driver
DATABASE_URL = "sqlite+aiosqlite:///./blog.db"The +aiosqlite part tells SQLAlchemy which async driver to use for the SQLite database. This is the critical change enabling async database operations.
Async Engine
engine = create_async_engine(DATABASE_URL)Use create_async_engine instead of create_engine. The check_same_thread argument is no longer needed—that was SQLite-specific for synchronous connections.
Async Session Factory
AsyncSessionLocal = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False
)Renamed to AsyncSessionLocal for clarity. Uses async_sessionmaker instead of sessionmaker. The class_=AsyncSession parameter specifies the session class.
expire_on_commit=False is recommended for async because it prevents issues with expired objects after commit. When an object expires, SQLAlchemy would normally try to reload it lazily, but lazy loading doesn’t work in async (explained in detail shortly).
Async Dependency Function
async def get_db():
async with AsyncSessionLocal() as session:
yield sessionThis is now an async function. Uses async with instead of with. Still a generator that yields a session, but now it’s an async generator.
Updating Main Application Imports
Add several new imports to main.py:
from contextlib import asynccontextmanager
from fastapi.exception_handlers import http_exception_handler, request_validation_exception_handler
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinloadBreaking these down:
asynccontextmanager— for the lifespan function (explained next)http_exception_handler,request_validation_error_handler— FastAPI’s default async exception handlersAsyncSession— replacesSessionfromsqlalchemy.ormselectinload— critical for eager loading relationships in asyncselect— already imported, but now used withselectinload
Remove JSONResponse from imports—it’s no longer needed as the default handlers are used instead.
Handling Database Table Creation with Lifespan
The current code uses Base.metadata.create_all(bind=engine) directly. This is synchronous and can’t be called with an async engine. Remove this line and create tables in a lifespan function instead.
Lifespan is the modern FastAPI way to handle startup and shutdown events. It replaces the older (now deprecated) @app.on_event("startup") and @app.on_event("shutdown") decorators seen in older tutorials.
@asynccontextmanager
async def lifespan(_app: FastAPI):
# Startup: create tables
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# Shutdown: dispose engine
await engine.dispose()
app = FastAPI(lifespan=lifespan)How this works:
- The
@asynccontextmanagerdecorator turns this into an async context manager. Code beforeyieldruns at startup.engine.begin()gets an async connection.run_sync()lets us run the synchronouscreate_all()method inside an async context. Theyieldis where the application actually runs. Code afteryieldruns at shutdown, disposing of the engine properly. - Pass the lifespan function to the FastAPI instance with
lifespan=lifespan. - This is the asynchronous equivalent of the previous table creation code—idempotent (safe to run multiple times without side effects).
The Critical Difference: Lazy Loading vs Eager Loading
This is probably the biggest difference between synchronous and asynchronous SQLAlchemy, and it trips up many developers.
Synchronous: Lazy Loading Works
In synchronous SQLAlchemy, lazy loading just works. When you have a post object and access post.author, SQLAlchemy automatically runs a query to load that author (it’s a relationship). Templates can access post.author.username without issues. This is called lazy loading.
Asynchronous: Lazy Loading Not Supported
In async SQLAlchemy, lazy loading is not supported. If you try to access post.author without having explicitly loaded it, you get an error like “missing greenlet” or similar. This happens because lazy loading would require running a synchronous query in an async context, which isn’t allowed.
Solution: Eager Loading with selectinload
Instead of letting SQLAlchemy lazy load relationships when you access them, explicitly tell SQLAlchemy to load them immediately with the main query:
result = await db.execute(
select(models.Post).options(selectinload(models.Post.author))
)When eager loading is required:
- Template routes that display
post.authordata - API routes returning
PostResponsewhich includes the author - Any query where relationships will be accessed
Without eager loading, accessing relationships in async contexts fails.
Converting Routes to Async
The pattern for converting routes is consistent. Demonstrate with the home route first:
Before (Synchronous):
@app.get("/", include_in_schema=False, name="home")
def home(
request: Request,
db: Annotated[Session, Depends(get_db)]
):
result = db.execute(select(models.Post))
posts = result.scalars().all()
return templates.TemplateResponse(
request,
"home.html",
{"posts": posts, "title": "Home"}
)After (Asynchronous):
@app.get("/", include_in_schema=False, name="home")
async def home(
request: Request,
db: Annotated[AsyncSession, Depends(get_db)]
):
result = await db.execute(
select(models.Post).options(selectinload(models.Post.author))
)
posts = result.scalars().all()
return templates.TemplateResponse(
request,
"home.html",
{"posts": posts, "title": "Home"}
)Changes made:
def→async defSession→AsyncSessionawait db.execute(...).options(selectinload(models.Post.author))for eager loading
When the template iterates over posts and accesses post.author, it works because the data was already loaded.
Post Page Route
@app.get("/posts/{post_id}", include_in_schema=False)
async def post_page(
request: Request,
post_id: int,
db: Annotated[AsyncSession, Depends(get_db)]
):
result = await db.execute(
select(models.Post)
.options(selectinload(models.Post.author))
.where(models.Post.id == post_id)
)
post = result.scalars().first()
# ... rest of the functionNote the order: .options(selectinload(...)) comes before .where(...). Both async conversion and eager loading applied.
Pattern Summary for Route Conversion
Every route using the database follows this pattern:
def→async defSession→AsyncSessionawait db.execute(...)- Add
selectinloadwhen relationships are accessed
When selectinload is NOT Needed
If a query doesn’t access relationships, selectinload isn’t needed:
# Checking if user exists - not accessing relationships
result = await db.execute(
select(models.User).where(models.User.id == user_id)
)
user = result.scalars().first()But the next query for posts DOES need it:
# Getting user's posts - template accesses post.author
result = await db.execute(
select(models.Post)
.options(selectinload(models.Post.author))
.where(models.Post.user_id == user_id)
)You must know which relationships your database has and what queries access them.
Awaiting Database Operations
Not all database methods require await:
Requires await:
db.execute()— performs I/Odb.commit()— writes to databasedb.refresh()— reloads from databasedb.delete()— interacts with session in a way requiring async
Does NOT require await:
db.add()— just adds the object to the session’s pending list in memory; no I/O happens
Example from create user:
new_user = models.User(username=user.username, email=user.email)
db.add(new_user) # No await
await db.commit() # Await required
await db.refresh(new_user) # Await requiredThe actual database operations happen at commit() and refresh()—those need await. add() does not.
Eager Loading in Refresh
When creating or updating resources that return relationships, use attribute_names in refresh():
@app.post("/api/posts", response_model=PostResponse, status_code=status.HTTP_201_CREATED)
async def create_post(
post: PostCreate,
db: Annotated[AsyncSession, Depends(get_db)]
):
# ... validation and creation logic ...
new_post = models.Post(
title=post.title,
content=post.content,
user_id=post.user_id
)
db.add(new_post)
await db.commit()
await db.refresh(new_post, attribute_names=["author"])
return new_postWhen creating a new post and returning it, the PostResponse schema includes the author. Instead of doing a separate query with selectinload, tell refresh() to also load specific relationships using the attribute_names parameter. await db.refresh(new_post, attribute_names=["author"]) refreshes the post and loads the author relationship in one operation.
This same pattern is used in update_post_full and update_post_partial.
Converting Exception Handlers to Async
Current handlers are synchronous and manually create JSON responses. A better approach uses FastAPI’s default handlers which are async, providing consistent behavior with the rest of FastAPI and requiring less code to maintain.
Before (Synchronous):
@app.exception_handler(StarletteHTTPException)
def http_exception_handler(request: Request, exc: StarletteHTTPException):
message = exc.detail if exc.detail else "An error occurred."
if request.url.path.startswith("/api"):
return JSONResponse(
status_code=exc.status_code,
content={"detail": message}
)
return templates.TemplateResponse(
request,
"error.html",
{"status_code": exc.status_code, "message": message, "title": exc.status_code},
status_code=exc.status_code
)After (Asynchronous):
@app.exception_handler(StarletteHTTPException)
async def general_http_exception_handler(request: Request, exception: StarletteHTTPException):
if request.url.path.startswith("/api"):
return await http_exception_handler(request, exception)
message = (
exception.detail
if exception.detail
else "An error occurred. Please check your request and try again."
)
return templates.TemplateResponse(
request,
"error.html",
{
"status_code": exception.status_code,
"title": exception.status_code,
"message": message,
},
status_code=exception.status_code,
)Changes:
- Function is now
async - For API routes,
awaitFastAPI’s defaulthttp_exception_handler(request, exc) - Message computation moved—only needed for template routes now
Same changes for validation exception handler:
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exception: RequestValidationError):
if request.url.path.startswith("/api"):
return await request_validation_exception_handler(request, exception)
return templates.TemplateResponse(
request,
"error.html",
{
"status_code": status.HTTP_422_UNPROCESSABLE_CONTENT,
"title": status.HTTP_422_UNPROCESSABLE_CONTENT,
"message": "Invalid request. Please check your input and try again.",
},
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
)Testing the Async Application
Start the development server. If you get a “no module named greenlet” error, install it:
uv add greenletOnce the server starts successfully, test:
- Homepage loads and displays posts
- Individual post pages work
- Documentation at
/docsfunctions - Create a new post through the docs
Create a test post:
{
"title": "Testing Async",
"content": "Testing async content",
"user_id": 1
}Execute successfully. Reload the homepage—the new post appears.
Critical verification: Get all posts from the API. Each post includes full author information (username, email, image path, etc.). This proves eager loading is working—the selectinload calls and attribute_names in refreshes are functioning correctly.
Externally, everything works exactly as before. All changes are internal—the application now uses async operations, enabling it to handle more concurrent requests efficiently.
When to Use Async
Use async when:
- Making multiple independent I/O operations
- Calling external APIs (probably the best use case)
- Database operations under high concurrency
- Long-running I/O operations
Use regular synchronous functions when:
- Simple, fast operations
- CPU work like calculations or image processing
- Code clarity matters more than marginal performance gains
- Using sync-only libraries
Critical point: You don’t have to choose one or the other. A mixed approach is totally fine and actually common. Some routes can be async, others synchronous. FastAPI handles both gracefully. Choose based on each route’s specific needs.
All routes in this application talk to the database, so all were converted. But standalone routes without I/O could remain synchronous.
Common Pitfalls to Avoid
Don’t do blocking I/O in async functions:
- Don’t use synchronous database sessions in async routes
- Don’t use synchronous libraries like
requestsin async routes—usehttpxinstead
Don’t forget eager loading:
- Any relationship accessed in templates or responses needs
selectinloadorattribute_names
Don’t add await to everything:
- Only I/O operations need
await db.add()does not needawait
Performance Reality
The benefits of async really show up under load. For a single request at a time (like local testing), you won’t see much difference. Async shines with many concurrent requests.
For learning and development, either approach is fine. Don’t optimize prematurely. For production under load, async can handle more concurrent connections. But if asynchronous code seems intimidating, using synchronous routes is completely valid—deploy your application, monitor logs and performance, and optimize with async if needed. It’s always an option later.
Summary
The database was converted to use create_async_engine with the aiosqlite driver, async_sessionmaker replaced the session maker, and get_db became an async generator. A lifespan function was added for table creation at startup and engine disposal at shutdown. All routes are now asynchronous since they use the database, with await on database operations. Exception handlers use FastAPI’s default async handlers. Eager loading relationships requires selectinload in queries and attribute_names in refresh calls.
The conversion pattern is consistent: async def, AsyncSession, await db.execute(), and eager loading where relationships are accessed.
The next chapter reorganizes code with routers. The main.py file has grown substantially and will continue growing with more features. The code will be broken into organized modules using APIRouter for posts and users, similar to Flask blueprints. This professional code organization pattern prepares the application for continued development.