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 def routes: FastAPI automatically runs them in a separate thread pool. This prevents the function from blocking the main event loop. Even with regular def functions, other requests can still be processed. This is automatic and works well.
  • async def routes: FastAPI runs them directly in the main event loop. This is more efficient, but you must await any I/O operations. If you do blocking I/O without await, you’ll block the entire event loop—this is worse than just using a regular def function.

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 aiosqlite

aiosqlite 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 greenlet

Converting 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 DeclarativeBase

Instead 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 session

This 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 selectinload

Breaking these down:

  • asynccontextmanager — for the lifespan function (explained next)
  • http_exception_handler, request_validation_error_handler — FastAPI’s default async exception handlers
  • AsyncSession — replaces Session from sqlalchemy.orm
  • selectinload — critical for eager loading relationships in async
  • select — already imported, but now used with selectinload

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 @asynccontextmanager decorator turns this into an async context manager. Code before yield runs at startup. engine.begin() gets an async connection. run_sync() lets us run the synchronous create_all() method inside an async context. The yield is where the application actually runs. Code after yield runs 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.author data
  • API routes returning PostResponse which 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:

  1. defasync def
  2. SessionAsyncSession
  3. await db.execute(...)
  4. .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 function

Note 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:

  1. defasync def
  2. SessionAsyncSession
  3. await db.execute(...)
  4. Add selectinload when 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/O
  • db.commit() — writes to database
  • db.refresh() — reloads from database
  • db.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 required

The 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_post

When 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:

  1. Function is now async
  2. For API routes, await FastAPI’s default http_exception_handler(request, exc)
  3. 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 greenlet

Once the server starts successfully, test:

  • Homepage loads and displays posts
  • Individual post pages work
  • Documentation at /docs functions
  • 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 requests in async routes—use httpx instead

Don’t forget eager loading:

  • Any relationship accessed in templates or responses needs selectinload or attribute_names

Don’t add await to everything:

  • Only I/O operations need await
  • db.add() does not need await

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.