Python Academy · Lesson

Async Endpoints and Database Integration

Write async route handlers and connect to a database with SQLAlchemy.

Async Endpoints and Database Integration is a free Python Academy lesson on CoddyKit — lesson 4 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 Python Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

async def Endpoints

Declare route handlers with async def for non-blocking I/O. Use def for CPU-bound or blocking operations (FastAPI runs them in a thread pool).

from fastapi import FastAPI
import asyncio
app = FastAPI()

@app.get("/slow")
async def slow():
    await asyncio.sleep(1)   # non-blocking
    return {"done": True}

asyncpg for PostgreSQL

asyncpg is a fast, async PostgreSQL driver. Use a connection pool shared across requests.

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool("postgresql://user:pass@host/db")
    async with pool.acquire() as conn:
        rows = await conn.fetch("SELECT id, name FROM users")
    await pool.close()
    return rows

SQLAlchemy Async

SQLAlchemy 1.4+ with async_sessionmaker provides an async ORM. Use AsyncSession as a FastAPI dependency.

# pip install sqlalchemy aiosqlite
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

engine = create_async_engine("sqlite+aiosqlite:///db.sqlite")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

Database Dependency

Yield the session from a FastAPI dependency so it is committed and closed after the request.

from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI()

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

SQLAlchemy ORM Models

Define ORM models with DeclarativeBase and map them to database tables.

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"

    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str]
    email: Mapped[str]

CRUD Operations

Implement basic Create, Read, Update, Delete using SQLAlchemy async session methods.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def create_user(db: AsyncSession, name: str, email: str):
    user = User(name=name, email=email)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

async def get_user(db: AsyncSession, user_id: int):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

Alembic for Migrations

Use Alembic to manage database schema migrations in a version-controlled way.

# pip install alembic
# alembic init alembic
# Edit alembic/env.py to point to your models

# Generate a migration:
# alembic revision --autogenerate -m "add users table"

# Apply:
# alembic upgrade head

httpx for Async HTTP Calls

Use httpx.AsyncClient to call external APIs from async FastAPI endpoints.

# pip install httpx
import httpx
from fastapi import FastAPI
app = FastAPI()

@app.get("/weather")
async def weather(city: str):
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://wttr.in/{city}?format=j1")
        return r.json()

Background Tasks

Use BackgroundTasks to run work after the response is sent — e.g., send an email, update a cache.

from fastapi import FastAPI, BackgroundTasks
app = FastAPI()

def send_welcome_email(email: str):
    # blocking I/O — runs in background thread
    smtp_send(email, "Welcome!")

@app.post("/register")
def register(email: str, bg: BackgroundTasks):
    create_user(email)
    bg.add_task(send_welcome_email, email)
    return {"status": "registered"}

Lifespan Events

Use the lifespan context manager (FastAPI 0.93+) to run startup/shutdown code — e.g., create the DB pool.

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = await asyncpg.create_pool(DSN)
    yield
    await app.state.pool.close()

app = FastAPI(lifespan=lifespan)

Testing FastAPI with httpx

Use httpx.AsyncClient with ASGITransport to test FastAPI endpoints without running a server.

import pytest, httpx
from fastapi import FastAPI
app = FastAPI()

@app.get("/ping")
async def ping(): return {"pong": True}

@pytest.mark.asyncio
async def test_ping():
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        r = await client.get("/ping")
    assert r.json() == {"pong": True}

Quick Check

When should you use async def vs def for a FastAPI route handler?

Recap

Use async def endpoints with async database drivers (asyncpg, SQLAlchemy async). Manage the DB session as a FastAPI dependency. Use BackgroundTasks for post-response work and the lifespan context manager for startup/shutdown.

Frequently asked questions

Is the “Async Endpoints and Database Integration” lesson free?

Yes — the full text of “Async Endpoints and Database Integration” is free to read here on the web, and the Python Academy 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 Python Academy course, upgrade to CoddyKit PRO.

What will I learn in “Async Endpoints and Database Integration”?

Write async route handlers and connect to a database with SQLAlchemy. You practise Python Academy 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 Python Academy?

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

How long does the “Async Endpoints and Database Integration” 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 Python Academy lesson?

Yes. Every Python Academy 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

FastAPI Project Setup and First Endpoint
Path Parameters, Query Params, and Request Bodies
Dependency Injection and Authentication
Async Endpoints and Database Integration

← Back to Python Academy