FastAPI Backend Development Bootcamp · Lesson

Asynchronous Database Access

Explore asynchronous database drivers and ORMs like `asyncpg` and `SQLModel` for non-blocking database operations.

Asynchronous Database Access is a free FastAPI Backend Development Bootcamp lesson on CoddyKit — lesson 2 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the FastAPI Backend Development Bootcamp learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

Why Async DB Access?

When building high-performance web applications, especially with frameworks like FastAPI, database operations can often become a bottleneck. Traditional database calls are "blocking", meaning your application waits for the database to respond before moving on.

Asynchronous database access allows your application to perform other tasks while waiting for the database, preventing your API from becoming unresponsive under heavy load. This is crucial for scalability!

Sync vs. Async DB Calls

Imagine a restaurant where one chef handles everything. If a customer orders a complex dish (a database query), the chef stops all other work until that dish is complete. This is synchronous I/O.

In an asynchronous setup, the chef starts the complex dish, then immediately moves to prepare simpler dishes or take new orders while the complex dish cooks in the background. When the complex dish is ready, the chef picks it up. This non-blocking approach boosts efficiency!

Meet asyncpg: The Async Driver

For PostgreSQL, the go-to asynchronous driver in Python is asyncpg. It's a high-performance library specifically designed to work with Python's asyncio framework.

Fast: Written partly in C for speed.
Asynchronous: Fully non-blocking.
Direct: Provides a direct interface to PostgreSQL.

It's often used as the underlying driver for async ORMs or when you need fine-grained control.

asyncpg in Action

Let's see a basic example of connecting to a PostgreSQL database and running a simple query using asyncpg. Remember to replace placeholder credentials with your own!

import asyncio
import asyncpg

async def main():
    conn = None
    try:
        conn = await asyncpg.connect(user='postgres', password='mysecretpassword',
                                     database='testdb', host='localhost')
        print("Connected to PostgreSQL!")
        
        # Execute a query
        result = await conn.fetchval('SELECT 1 + 1')
        print(f"Query result: {result}") # Should be 2
        
    except Exception as e:
        print(f"Error: {e}")
    finally:
        if conn:
            await conn.close()
            print("Connection closed.")

if __name__ == "__main__":
    asyncio.run(main())

Awaiting Database Calls

In the previous example, you saw the await keyword before asyncpg.connect() and conn.fetchval(). This is crucial for asynchronous operations.

await tells Python: "This operation might take time, so pause here and let other tasks run in the meantime."
When the database operation completes, the task resumes from where it left off.
This non-blocking wait is what makes your FastAPI application scalable.

SQLModel: Async ORM Power

While asyncpg gives you low-level control, an Object Relational Mapper (ORM) simplifies database interactions by mapping database tables to Python objects. SQLModel is a modern, async-first ORM built on:

Pydantic: For data validation and serialization.
SQLAlchemy: The powerful and mature Python SQL toolkit.

It lets you define models once and use them for both your API request/response and database schema!

Setting up SQLModel for Async

To use SQLModel asynchronously, you need an asynchronous database engine. This typically involves using an async driver like asyncpg (which SQLAlchemy can use via asyncio). Here's how you'd set up the engine:

from sqlmodel import create_engine, SQLModel
import asyncio

# Replace with your actual async PostgreSQL connection string
# The 'postgresql+asyncpg' part tells SQLAlchemy to use asyncpg
DATABASE_URL = "postgresql+asyncpg://postgres:mysecretpassword@localhost/testdb"

async def main():
    engine = create_engine(DATABASE_URL, echo=True)
    print("Async SQLModel engine created.")
    
    # In a real app, you'd usually create tables here
    # async with engine.begin() as conn:
    #     await conn.run_sync(SQLModel.metadata.create_all)
    
    # Just demonstrating engine creation for this example
    await engine.dispose()
    print("Engine disposed.")

if __name__ == "__main__":
    asyncio.run(main())

Your First SQLModel

Defining a model in SQLModel is super intuitive. You inherit from SQLModel and use Pydantic-like field declarations. This single definition creates both your database table schema and your API's data validation schema!

from typing import Optional
from sqlmodel import Field, SQLModel

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

# This model can now be used with FastAPI for request bodies
# and with SQLAlchemy for database interactions.
print("Hero model defined successfully!")

Async CRUD with SQLModel

Now let's perform a simple Create and Read operation using our Hero model and the async engine. We'll use AsyncSession from SQLAlchemy's ORM for database interactions.

from typing import Optional
from sqlmodel import Field, SQLModel, create_engine, Session, select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
import asyncio

DATABASE_URL = "postgresql+asyncpg://postgres:mysecretpassword@localhost/testdb"
async_engine = create_async_engine(DATABASE_URL, echo=False)
AsyncSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=async_engine, class_=AsyncSession)

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

async def create_db_and_tables():
    async with async_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    print("Database tables created/updated.")

async def create_hero(hero: Hero):
    async with AsyncSessionLocal() as session:
        session.add(hero)
        await session.commit()
        await session.refresh(hero)
        print(f"Created hero: {hero.name} (ID: {hero.id})")
        return hero

async def get_heroes():
    async with AsyncSessionLocal() as session:
        statement = select(Hero)
        results = await session.execute(statement)
        heroes = results.scalars().all()
        print("\nAll Heroes:")
        for hero in heroes:
            print(f"- {hero.name} (Age: {hero.age})")
        return heroes

async def main():
    await create_db_and_tables()

    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson", age=28)
    hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
    hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)

    await create_hero(hero_1)
    await create_hero(hero_2)
    await create_hero(hero_3)

    await get_heroes()
    await async_engine.dispose()

if __name__ == "__main__":
    asyncio.run(main())

Async DB Check

You've learned about the importance of asynchronous database access and explored tools like asyncpg and SQLModel. Let's test your understanding!

Recap: Async DB for Scale

Great job! You've successfully explored asynchronous database access.

We understood why non-blocking I/O is vital for high-performance FastAPI apps.
We introduced asyncpg as a low-level async PostgreSQL driver.
We learned about SQLModel, an async-first ORM combining Pydantic and SQLAlchemy.
We saw practical examples of setting up and performing CRUD operations with these tools.

Mastering async database interactions is a key step towards building truly scalable and responsive backend services!

Frequently asked questions

Is the “Asynchronous Database Access” lesson free?

Yes — the full text of “Asynchronous Database Access” is free to read here on the web, and the FastAPI Backend Development Bootcamp course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the FastAPI Backend Development Bootcamp course, upgrade to CoddyKit PRO.

What will I learn in “Asynchronous Database Access”?

Explore asynchronous database drivers and ORMs like `asyncpg` and `SQLModel` for non-blocking database operations. You practise FastAPI Backend Development Bootcamp with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start FastAPI Backend Development Bootcamp?

No prior experience is required. FastAPI Backend Development Bootcamp on CoddyKit is structured for beginners through advanced learners; this is — lesson 2 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Asynchronous Database Access” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this FastAPI Backend Development Bootcamp lesson?

Yes. Every FastAPI Backend Development Bootcamp lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

← Back to FastAPI Backend Development Bootcamp