Python

Stop blocking a worker while the database thinks. Convert StudyDesk to async def routes, an async engine, and async sessions, and understand what actually changes.

Where StudyDesk is now

src/studydesk/
├── database.py          ← sync engine (psycopg), Session, get_db
├── crud.py              ← sync functions
├── routers/*.py         ← def routes
├── worker.py, tasks.py, seed.py
└── ...
tests/                   ← 63 sync tests

Everything is synchronous. By the end of this article, the engine is asyncpg-backed, sessions are AsyncSession, every route and CRUD function is async, the worker and seed are async, and the test suite runs the same 63 checks against the async app.

Part 1: The Idle CPU

Neha's load test: 200 simultaneous GET /courses/1. The API answers them all, but slowly, and the CPU graph sits near zero. The server is not working; it is waiting.

Each request does almost no computation. It sends a query to PostgreSQL, waits a few milliseconds for the answer, serialises it, and returns. During the wait, the thread handling that request can do nothing else. FastAPI runs plain def routes on a thread pool (40 threads by default), so at most 40 requests can be waiting at once; the other 160 queue up behind them.

"The waiting is the problem," says Neha. "Python can hand a thread something else to do while the database thinks. That's async."

Part 2: async and await, Explained

Blocking

def get_student(db, student_id):
    return db.get(Student, student_id)   # sends SQL, then BLOCKS until the reply arrives

While db.get waits for PostgreSQL, this thread is stuck. It cannot serve another request.

Non-blocking

async def get_student(db, student_id):
    return await db.get(Student, student_id)   # sends SQL, then YIELDS control until the reply arrives

await means: "this will take a while; let someone else run, and wake me up when it's done." The thread goes off to handle another request. When PostgreSQL replies, this function resumes exactly where it paused.

That pause-and-resume should sound familiar: it is the same mechanism as a generator's yield from Article 7. An async def function is a coroutine, a function that can be paused at every await and resumed later.

The event loop

Something has to decide who runs next. That is the event loop: a scheduler that keeps a list of paused coroutines and runs whichever one is ready. Uvicorn runs an event loop; FastAPI hands it your async def routes. One thread, thousands of in-flight requests, as long as each one spends its time awaiting rather than computing.

The two rules

await only works inside async def.
You can only await things that are awaitable: other async def functions, and async library calls (await db.get(...), await asyncio.sleep(1)). Awaiting a plain function is a TypeError; calling an async function without await runs nothing and returns an unstarted coroutine object (a very common bug).

What async does not do

Async helps when the time is spent waiting on I/O: databases, HTTP calls, files, sleeps. It does not help with CPU work; a 5-second calculation still takes 5 seconds and blocks the loop while it runs. And it is not magic parallelism: one event loop runs one coroutine at a time, just never sits idle.

The one thing that must be true all the way down

If a route is async def but calls a blocking function inside (a sync database driver, time.sleep, requests.get), the whole event loop freezes for that call, and every other request waits. Async only works if every wait is an await. That is why we cannot just add async to the routes: the database layer must become async too.

Part 3: What Changes, in One Table

Piece	Before	After
Driver	`psycopg` (sync)	`asyncpg` (async)
URL	`postgresql+psycopg://...`	`postgresql+asyncpg://...`
Engine	`create_engine`	`create_async_engine`
Session	`Session`, `sessionmaker`	`AsyncSession`, `async_sessionmaker`
`get_db`	generator	async generator
CRUD	`def`, `db.get(...)`	`async def`, `await db.get(...)`
Routes	`def`	`async def` + `await crud...`
`get_current_user`	`def`	`async def`
Worker loop	`time.sleep`	`asyncio.sleep`
Tests	`TestClient`	`httpx.AsyncClient` + `pytest-asyncio`
Alembic	unchanged (uses a sync URL)	unchanged

Install:

uv add asyncpg
uv add --dev pytest-asyncio

We keep psycopg installed: Alembic will continue to use it.

Update the URLs

.env — change the two database lines:

DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/studydesk
TEST_DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/studydesk_test

.env.example — same change.

docker-compose.yml — in migrate, api, and worker, change the DATABASE_URL value to:

      DATABASE_URL: postgresql+asyncpg://postgres:postgres@db:5432/studydesk

(All three services; nothing else in the file changes.)

Part 4: Async Engine and Session

src/studydesk/database.py

from collections.abc import AsyncGenerator
from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from .config import DATABASE_URL

engine = create_async_engine(DATABASE_URL, echo=False)

SessionLocal = async_sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)


class Base(DeclarativeBase):
    """Parent class for every SQLAlchemy model."""


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """Open an async database session for one request, and always close it."""
    async with SessionLocal() as session:
        yield session


DbSession = Annotated[AsyncSession, Depends(get_db)]

Reading it

create_async_engine, async_sessionmaker, AsyncSession The async versions. Same ideas: one engine with a pool, a factory that makes sessions.

expire_on_commit=False Important for async. By default, after commit() SQLAlchemy marks every loaded attribute as stale, and the next access (student.name) silently re-queries the database. In async code there is no "silent" query: an implicit load outside an await raises MissingGreenlet. Setting expire_on_commit=False keeps the values you already have, and we await db.refresh(...) explicitly when we want fresh ones.

async with SessionLocal() as session: yield session async with is the async form of with. The session is closed when the block ends. This is an async generator: async def plus yield. FastAPI handles it exactly like the sync one: runs to yield before the route, resumes after.

DbSession is now AsyncSession. Every route that says db: DbSession gets the async one.

Alembic keeps a sync connection

Alembic's env.py imports DATABASE_URL, which is now an asyncpg URL. Alembic can run async engines, but the simplest reliable approach is to give it the sync driver. One line changes in env.py:

alembic/env.py — replace the set_main_option line

config.set_main_option("sqlalchemy.url", DATABASE_URL.replace("+asyncpg", "+psycopg"))

Migrations run exactly as before. The rest of env.py is unchanged.

Part 5: Async CRUD

The mechanical rule for every function:

Sync	Async
`def f(db: Session, ...)`	`async def f(db: AsyncSession, ...)`
`db.get(Model, id)`	`await db.get(Model, id)`
`db.scalars(stmt).first()`	`await db.scalar(stmt)`
`db.scalars(stmt).all()`	`(await db.scalars(stmt)).all()`
`db.scalar(stmt)`	`await db.scalar(stmt)`
`db.commit()` / `db.rollback()` / `db.refresh()`	`await db.commit()` / `await db.rollback()` / `await db.refresh()`
`db.add(obj)` / `db.delete(obj)`	`db.add(obj)` (no await) / `await db.delete(obj)`

db.add only stages an object in memory, so it stays synchronous. db.delete may need to load relationships for cascades, so it is awaited.

src/studydesk/crud.py

from datetime import datetime, timezone

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from .models import Course, Enrollment, Job, Student, User
from .schemas import CourseCreate, StudentCreate, UserCreate
from .security import hash_password, verify_password

GRADE_RANGES = {
    "A": (85, 100),
    "B": (70, 84),
    "C": (50, 69),
    "D": (0, 49),
}


class EnrollmentError(Exception):
    """Raised when an enrollment cannot be made. The message says why."""


# ---------- students ----------


async def get_student(db: AsyncSession, student_id: int) -> Student | None:
    return await db.get(Student, student_id)


async def get_student_by_email(db: AsyncSession, email: str) -> Student | None:
    stmt = select(Student).where(func.lower(Student.email) == email.lower())
    return await db.scalar(stmt)


async def list_students(
    db: AsyncSession,
    *,
    city: str | None = None,
    grade: str | None = None,
    min_marks: int | None = None,
    sort_by: str = "id",
    order: str = "asc",
    offset: int = 0,
    limit: int = 10,
) -> tuple[list[Student], int]:
    stmt = select(Student)

    if city:
        stmt = stmt.where(func.lower(Student.city) == city.lower())

    if grade:
        low, high = GRADE_RANGES[grade.upper()]
        stmt = stmt.where(Student.marks.between(low, high))

    if min_marks is not None:
        stmt = stmt.where(Student.marks >= min_marks)

    total = await db.scalar(select(func.count()).select_from(stmt.subquery()))

    column = getattr(Student, sort_by)
    stmt = stmt.order_by(column.desc() if order == "desc" else column.asc())
    stmt = stmt.offset(offset).limit(limit)

    rows = (await db.scalars(stmt)).all()
    return list(rows), total or 0


async def create_student(db: AsyncSession, payload: StudentCreate) -> Student:
    student = Student(**payload.model_dump())
    db.add(student)
    await db.commit()
    await db.refresh(student)
    return student


async def update_student(db: AsyncSession, student: Student, changes: dict) -> Student:
    for field, value in changes.items():
        setattr(student, field, value)
    await db.commit()
    await db.refresh(student)
    return student


async def delete_student(db: AsyncSession, student: Student) -> None:
    """Remove a student; free the seat in every course they were enrolled in."""
    enrollments = (
        await db.scalars(select(Enrollment).where(Enrollment.student_id == student.id))
    ).all()
    for enrollment in enrollments:
        course = await db.scalar(
            select(Course).where(Course.id == enrollment.course_id).with_for_update()
        )
        if course is not None and course.seats_taken > 0:
            course.seats_taken -= 1
    await db.delete(student)
    await db.commit()


async def list_student_courses(db: AsyncSession, student_id: int) -> list[Enrollment]:
    stmt = (
        select(Enrollment)
        .options(selectinload(Enrollment.course))
        .where(Enrollment.student_id == student_id)
        .order_by(Enrollment.enrolled_at)
    )
    return list((await db.scalars(stmt)).all())


# ---------- courses ----------


async def get_course(db: AsyncSession, course_id: int) -> Course | None:
    return await db.get(Course, course_id)


async def get_course_with_students(db: AsyncSession, course_id: int) -> Course | None:
    stmt = (
        select(Course)
        .options(selectinload(Course.enrollments).selectinload(Enrollment.student))
        .where(Course.id == course_id)
    )
    return await db.scalar(stmt)


async def get_course_by_code(db: AsyncSession, code: str) -> Course | None:
    return await db.scalar(select(Course).where(Course.code == code.upper()))


async def list_courses(db: AsyncSession) -> list[Course]:
    return list((await db.scalars(select(Course).order_by(Course.code))).all())


async def create_course(db: AsyncSession, payload: CourseCreate) -> Course:
    course = Course(code=payload.code.upper(), title=payload.title, seats=payload.seats)
    db.add(course)
    await db.commit()
    await db.refresh(course)
    return course


# ---------- enrollments (transactional) ----------


async def enroll_student(db: AsyncSession, course_id: int, student_id: int) -> Enrollment:
    try:
        course = await db.scalar(select(Course).where(Course.id == course_id).with_for_update())
        if course is None:
            raise EnrollmentError(f"Course {course_id} not found")

        student = await db.get(Student, student_id)
        if student is None:
            raise EnrollmentError(f"Student {student_id} not found")

        already = await db.scalar(
            select(Enrollment).where(
                Enrollment.course_id == course_id,
                Enrollment.student_id == student_id,
            )
        )
        if already is not None:
            raise EnrollmentError(f"Student {student_id} is already enrolled in {course.code}")

        if course.seats_taken >= course.seats:
            raise EnrollmentError(f"Course {course.code} is full ({course.seats} seats)")

        enrollment = Enrollment(student_id=student_id, course_id=course_id)
        db.add(enrollment)
        course.seats_taken += 1

        await db.commit()
    except Exception:
        await db.rollback()
        raise

    await db.refresh(enrollment, attribute_names=["id", "enrolled_at", "student"])
    return enrollment


async def unenroll_student(db: AsyncSession, course_id: int, student_id: int) -> bool:
    try:
        course = await db.scalar(select(Course).where(Course.id == course_id).with_for_update())
        if course is None:
            return False

        enrollment = await db.scalar(
            select(Enrollment).where(
                Enrollment.course_id == course_id,
                Enrollment.student_id == student_id,
            )
        )
        if enrollment is None:
            await db.rollback()
            return False

        await db.delete(enrollment)
        course.seats_taken -= 1
        await db.commit()
        return True
    except Exception:
        await db.rollback()
        raise


# ---------- users ----------


async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    return await db.scalar(select(User).where(func.lower(User.email) == email.lower()))


async def create_user(db: AsyncSession, payload: UserCreate, role: str = "staff") -> User:
    user = User(
        email=payload.email.lower(),
        hashed_password=hash_password(payload.password),
        role=role,
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user


async def authenticate_user(db: AsyncSession, email: str, password: str) -> User | None:
    user = await get_user_by_email(db, email)
    if user is None:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user


# ---------- jobs ----------


async def create_job(db: AsyncSession, kind: str, payload: dict | None = None) -> Job:
    job = Job(kind=kind, payload=payload or {})
    db.add(job)
    await db.commit()
    await db.refresh(job)
    return job


async def get_job(db: AsyncSession, job_id: int) -> Job | None:
    return await db.get(Job, job_id)


async def claim_next_job(db: AsyncSession) -> Job | None:
    stmt = (
        select(Job)
        .where(Job.status == "pending")
        .order_by(Job.created_at)
        .limit(1)
        .with_for_update(skip_locked=True)
    )
    job = await db.scalar(stmt)
    if job is None:
        await db.rollback()
        return None

    job.status = "running"
    job.attempts += 1
    job.started_at = datetime.now(timezone.utc)
    await db.commit()
    await db.refresh(job)
    return job


async def finish_job(db: AsyncSession, job: Job, result: dict) -> None:
    job.status = "done"
    job.result = result
    job.error = None
    job.finished_at = datetime.now(timezone.utc)
    await db.commit()


async def fail_job(db: AsyncSession, job: Job, error: str) -> None:
    job.error = error
    job.finished_at = datetime.now(timezone.utc)
    job.status = "pending" if job.attempts < job.max_attempts else "failed"
    await db.commit()

Two things beyond the mechanical rule

delete_student now frees seats The fix from Article 14's quiz, done properly: load the student's enrollments, lock and decrement each course, then delete. All inside one transaction. (The cascade still removes the enrollment rows.)

await db.refresh(enrollment, attribute_names=["id", "enrolled_at", "student"]) In async, you cannot rely on a lazy load of enrollment.student during response serialisation; it would raise. So after the commit we refresh exactly the attributes the response needs, including the relationship, in one explicit query.

Part 6: Async Dependencies and Routers

src/studydesk/dependencies.py

from typing import Annotated

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from . import crud
from .database import DbSession
from .models import User
from .security import decode_access_token

bearer_scheme = HTTPBearer(auto_error=False)


async def get_current_user(
    db: DbSession,
    credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer_scheme)],
) -> User:
    credentials_error = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    if credentials is None:
        raise credentials_error

    payload = decode_access_token(credentials.credentials)
    if payload is None:
        raise credentials_error

    email = payload.get("sub")
    if email is None:
        raise credentials_error

    user = await crud.get_user_by_email(db, email)
    if user is None or not user.is_active:
        raise credentials_error

    return user


CurrentUser = Annotated[User, Depends(get_current_user)]


def require_admin(user: CurrentUser) -> User:
    if user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin privileges required",
        )
    return user


AdminUser = Annotated[User, Depends(require_admin)]

require_admin stays a plain def: it does no I/O. FastAPI happily mixes sync and async dependencies.

src/studydesk/routers/auth.py

from fastapi import APIRouter, HTTPException, status

from .. import crud
from ..database import DbSession
from ..dependencies import CurrentUser
from ..schemas import LoginRequest, Token, UserCreate, UserOut
from ..security import create_access_token

router = APIRouter(prefix="/auth", tags=["Auth"])


@router.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED, summary="Register a new staff user")
async def register(payload: UserCreate, db: DbSession):
    if await crud.get_user_by_email(db, payload.email) is not None:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="A user with this email already exists")
    return await crud.create_user(db, payload, role="staff")


@router.post("/login", response_model=Token, summary="Log in and receive an access token")
async def login(payload: LoginRequest, db: DbSession):
    user = await crud.authenticate_user(db, payload.email, payload.password)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if not user.is_active:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="This account is disabled")

    token = create_access_token(subject=user.email, role=user.role)
    return Token(access_token=token)


@router.get("/me", response_model=UserOut, summary="Get the current user")
async def me(user: CurrentUser):
    return user

src/studydesk/routers/students.py

from typing import Annotated, Literal

from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Path, Query, status
from sqlalchemy.exc import IntegrityError

from .. import crud
from ..database import DbSession
from ..dependencies import AdminUser, CurrentUser
from ..schemas import StudentCourseOut, StudentCreate, StudentListOut, StudentOut, StudentUpdate
from ..tasks import send_welcome_email

router = APIRouter(prefix="/students", tags=["Students"])

StudentId = Annotated[
    int,
    Path(title="Student ID", description="The unique ID of the student. Must be a positive integer.", gt=0, examples=[1]),
]


async def get_student_or_404(db: DbSession, student_id: StudentId):
    student = await crud.get_student(db, student_id)
    if student is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Student with id {student_id} not found")
    return student


@router.get("", response_model=StudentListOut, summary="Get all students")
async def list_students(
    db: DbSession,
    user: CurrentUser,
    city: Annotated[str | None, Query(description="Filter by city (case-insensitive).", examples=["Ahmedabad"])] = None,
    grade: Annotated[str | None, Query(description="Filter by grade: A, B, C, or D.", pattern="^[ABCDabcd]$", examples=["A"])] = None,
    min_marks: Annotated[int | None, Query(description="Only students with marks >= this value.", ge=0, le=100)] = None,
    sort_by: Annotated[Literal["id", "name", "age", "marks"], Query(description="Field to sort by.")] = "id",
    order: Annotated[Literal["asc", "desc"], Query(description="Sort direction: asc or desc.")] = "asc",
    offset: Annotated[int, Query(description="Number of students to skip.", ge=0)] = 0,
    limit: Annotated[int, Query(description="Maximum number of students to return (1-100).", ge=1, le=100)] = 10,
):
    students, total = await crud.list_students(
        db, city=city, grade=grade, min_marks=min_marks, sort_by=sort_by, order=order, offset=offset, limit=limit
    )
    total_pages = (total + limit - 1) // limit if total > 0 else 0
    return StudentListOut(total=total, total_pages=total_pages, offset=offset, limit=limit, count=len(students), students=students)


@router.get("/{student_id}", response_model=StudentOut, summary="Get one student by ID")
async def get_student(user: CurrentUser, student=Depends(get_student_or_404)):
    return student


@router.get("/{student_id}/courses", response_model=list[StudentCourseOut], summary="List the courses a student is enrolled in")
async def get_student_courses(db: DbSession, user: CurrentUser, student=Depends(get_student_or_404)):
    return await crud.list_student_courses(db, student.id)


@router.post("", response_model=StudentOut, status_code=status.HTTP_201_CREATED, summary="Create a new student")
async def create_student(payload: StudentCreate, db: DbSession, user: CurrentUser, background_tasks: BackgroundTasks):
    if await crud.get_student_by_email(db, payload.email) is not None:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=f"A student with email {payload.email} already exists")

    try:
        student = await crud.create_student(db, payload)
    except IntegrityError:
        await db.rollback()
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=f"A student with email {payload.email} already exists")

    background_tasks.add_task(send_welcome_email, student.email, student.name)
    return student


@router.put("/{student_id}", response_model=StudentOut, summary="Update a student (partial)")
async def update_student(payload: StudentUpdate, db: DbSession, user: CurrentUser, student=Depends(get_student_or_404)):
    changes = payload.model_dump(exclude_unset=True)
    if not changes:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No fields provided to update")

    if "email" in changes:
        other = await crud.get_student_by_email(db, changes["email"])
        if other is not None and other.id != student.id:
            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=f"A student with email {changes['email']} already exists")

    try:
        return await crud.update_student(db, student, changes)
    except IntegrityError:
        await db.rollback()
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=f"A student with email {changes.get('email')} already exists")


@router.delete("/{student_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a student (admin only)")
async def delete_student(db: DbSession, admin: AdminUser, student=Depends(get_student_or_404)):
    await crud.delete_student(db, student)

send_welcome_email stays synchronous. BackgroundTasks runs plain functions on a thread pool, so its time.sleep(2) does not block the event loop. Only the request path needs to be async.

src/studydesk/routers/courses.py

from typing import Annotated

from fastapi import APIRouter, HTTPException, Path, status

from .. import crud
from ..database import DbSession
from ..dependencies import AdminUser, CurrentUser
from ..schemas import CourseCreate, CourseOut, CourseWithStudentsOut, EnrollmentOut

router = APIRouter(prefix="/courses", tags=["Courses"])

CourseId = Annotated[int, Path(gt=0, description="Course ID")]
StudentIdPath = Annotated[int, Path(gt=0, description="Student ID")]


@router.get("", response_model=list[CourseOut], summary="List all courses")
async def list_courses(db: DbSession, user: CurrentUser):
    return await crud.list_courses(db)


@router.post("", response_model=CourseOut, status_code=status.HTTP_201_CREATED, summary="Create a course (admin only)")
async def create_course(payload: CourseCreate, db: DbSession, admin: AdminUser):
    if await crud.get_course_by_code(db, payload.code) is not None:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=f"A course with code {payload.code.upper()} already exists")
    return await crud.create_course(db, payload)


@router.get("/{course_id}", response_model=CourseWithStudentsOut, summary="Get a course with its students")
async def get_course(course_id: CourseId, db: DbSession, user: CurrentUser):
    course = await crud.get_course_with_students(db, course_id)
    if course is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Course {course_id} not found")
    return course


@router.post("/{course_id}/students/{student_id}", response_model=EnrollmentOut, status_code=status.HTTP_201_CREATED, summary="Enrol a student in a course")
async def enroll(course_id: CourseId, student_id: StudentIdPath, db: DbSession, user: CurrentUser):
    try:
        return await crud.enroll_student(db, course_id, student_id)
    except crud.EnrollmentError as exc:
        message = str(exc)
        if "not found" in message:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=message)
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message)


@router.delete("/{course_id}/students/{student_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Remove a student from a course")
async def unenroll(course_id: CourseId, student_id: StudentIdPath, db: DbSession, user: CurrentUser):
    removed = await crud.unenroll_student(db, course_id, student_id)
    if not removed:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Student {student_id} is not enrolled in course {course_id}")

src/studydesk/routers/jobs.py

from typing import Annotated

from fastapi import APIRouter, HTTPException, Path, Query, status

from .. import crud
from ..database import DbSession
from ..dependencies import AdminUser
from ..schemas import JobOut

router = APIRouter(prefix="/jobs", tags=["Jobs"])


@router.post("/grade-report", response_model=JobOut, status_code=status.HTTP_202_ACCEPTED, summary="Queue a grade report (admin only)")
async def queue_grade_report(
    db: DbSession,
    admin: AdminUser,
    city: Annotated[str | None, Query(description="Limit the report to one city.")] = None,
):
    payload = {"city": city} if city else {}
    return await crud.create_job(db, kind="grade_report", payload=payload)


@router.get("/{job_id}", response_model=JobOut, summary="Get a job's status and result (admin only)")
async def get_job(db: DbSession, admin: AdminUser, job_id: Annotated[int, Path(gt=0)]):
    job = await crud.get_job(db, job_id)
    if job is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Job {job_id} not found")
    return job

src/studydesk/__init__.py — only db_check changes:

import logging
from contextlib import asynccontextmanager

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import func, select

from .config import ALLOWED_ORIGINS, HOST, PORT, RELOAD
from .database import DbSession
from .models import Student
from .routers import auth, courses, jobs, students

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield


app = FastAPI(
    title="StudyDesk API",
    description="Student records API for a coaching institute.",
    version="0.7.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(auth.router)
app.include_router(students.router)
app.include_router(courses.router)
app.include_router(jobs.router)


@app.get("/", tags=["General"])
async def home():
    return {"message": "Welcome to StudyDesk API"}


@app.get("/db-check", tags=["General"], summary="Check the database connection")
async def db_check(db: DbSession):
    total = await db.scalar(select(func.count()).select_from(Student))
    return {"database": "ok", "students_in_db": total}


def main() -> None:
    uvicorn.run("studydesk:app", host=HOST, port=PORT, reload=RELOAD)

Part 7: Async Tasks, Worker, and Seed

src/studydesk/tasks.py

import asyncio
import logging
import time

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

from .crud import GRADE_RANGES
from .models import Student

logger = logging.getLogger("studydesk.tasks")


def send_welcome_email(to_email: str, student_name: str) -> None:
    """Sync on purpose: BackgroundTasks runs it on a thread, so the sleep does not block the loop."""
    logger.info("Sending welcome email to %s ...", to_email)
    time.sleep(2)
    logger.info("Welcome email sent to %s (Dear %s, welcome to the institute!)", to_email, student_name)


async def generate_grade_report(db: AsyncSession, payload: dict) -> dict:
    city = payload.get("city")
    logger.info("Generating grade report (city=%s) ...", city)
    await asyncio.sleep(5)  # simulate heavy I/O-bound work without blocking the loop

    report: dict[str, int] = {}
    for grade, (low, high) in GRADE_RANGES.items():
        stmt = select(func.count()).select_from(Student).where(Student.marks.between(low, high))
        if city:
            stmt = stmt.where(func.lower(Student.city) == city.lower())
        report[grade] = await db.scalar(stmt) or 0

    total = sum(report.values())
    logger.info("Grade report ready: %s (total %d)", report, total)
    return {"city": city, "total": total, "by_grade": report}


TASKS = {
    "grade_report": generate_grade_report,
}

src/studydesk/worker.py

import asyncio
import logging

from . import crud
from .database import SessionLocal
from .tasks import TASKS

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
logger = logging.getLogger("studydesk.worker")

POLL_SECONDS = 2


async def run_one(db) -> bool:
    job = await crud.claim_next_job(db)
    if job is None:
        return False

    logger.info("Job %d (%s) started, attempt %d/%d", job.id, job.kind, job.attempts, job.max_attempts)
    task = TASKS.get(job.kind)

    if task is None:
        await crud.fail_job(db, job, f"unknown job kind: {job.kind}")
        logger.error("Job %d failed: unknown kind %s", job.id, job.kind)
        return True

    try:
        result = await task(db, job.payload)
        await crud.finish_job(db, job, result)
        logger.info("Job %d done", job.id)
    except Exception as exc:
        await db.rollback()
        await crud.fail_job(db, job, repr(exc))
        logger.error("Job %d failed: %s (status now %s)", job.id, exc, job.status)

    return True


async def loop() -> None:
    logger.info("Worker started, polling every %ds", POLL_SECONDS)
    while True:
        async with SessionLocal() as db:
            ran = await run_one(db)
        if not ran:
            await asyncio.sleep(POLL_SECONDS)


def main() -> None:
    asyncio.run(loop())


if __name__ == "__main__":
    main()

asyncio.run(loop()) — the entry point of any async program. It creates an event loop, runs the coroutine until it finishes, and closes the loop. main() stays a plain def because [project.scripts] expects a normal callable.

src/studydesk/seed.py

import asyncio
import json
from pathlib import Path

from sqlalchemy import func, select

from .config import ADMIN_EMAIL, ADMIN_PASSWORD
from .database import SessionLocal
from .models import Student, User
from .security import hash_password

DATA_FILE = Path(__file__).parent / "data" / "students.json"


async def seed_students(db) -> None:
    count = await db.scalar(select(func.count()).select_from(Student))
    if count:
        print("students: table already has data, skipping.")
        return

    with open(DATA_FILE, "r", encoding="utf-8") as f:
        records = json.load(f)

    for record in records:
        record.pop("grade", None)
        db.add(Student(**record))
    await db.commit()
    print(f"students: inserted {len(records)}.")


async def seed_admin(db) -> None:
    if not ADMIN_EMAIL or not ADMIN_PASSWORD:
        print("admin: ADMIN_EMAIL / ADMIN_PASSWORD not set, skipping.")
        return

    existing = await db.scalar(select(User).where(User.email == ADMIN_EMAIL.lower()))
    if existing is not None:
        print("admin: already exists, skipping.")
        return

    db.add(User(email=ADMIN_EMAIL.lower(), hashed_password=hash_password(ADMIN_PASSWORD), role="admin"))
    await db.commit()
    print(f"admin: created {ADMIN_EMAIL}.")


async def seed() -> None:
    async with SessionLocal() as db:
        await seed_students(db)
        await seed_admin(db)


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

Part 8: Async Tests

Setup

pyproject.toml — update the pytest section:

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

asyncio_mode = "auto" lets pytest-asyncio run any async def test_* and any async def fixture without extra decorators.

The new `conftest.py`

tests/conftest.py

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from studydesk import app
from studydesk.config import TEST_DATABASE_URL
from studydesk.database import Base, get_db
from studydesk.models import User
from studydesk.security import create_access_token, hash_password

if not TEST_DATABASE_URL:
    raise RuntimeError("TEST_DATABASE_URL is not set in .env")

test_engine = create_async_engine(TEST_DATABASE_URL)
TestSessionLocal = async_sessionmaker(bind=test_engine, autoflush=False, expire_on_commit=False)


# ---------- database ----------


@pytest.fixture(scope="session", autouse=True)
async def create_test_tables():
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await test_engine.dispose()


@pytest.fixture
async def db() -> AsyncSession:
    """A session inside a transaction that is rolled back after each test."""
    async with test_engine.connect() as connection:
        transaction = await connection.begin()
        session = TestSessionLocal(bind=connection, join_transaction_mode="create_savepoint")
        try:
            yield session
        finally:
            await session.close()
            await transaction.rollback()


# ---------- client ----------


@pytest.fixture
async def client(db: AsyncSession) -> AsyncClient:
    async def override_get_db():
        yield db

    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as async_client:
        yield async_client
    app.dependency_overrides.clear()


# ---------- users and tokens ----------


async def make_user(db: AsyncSession, email: str, password: str, role: str) -> User:
    user = User(email=email, hashed_password=hash_password(password), role=role)
    db.add(user)
    await db.flush()
    return user


@pytest.fixture
async def staff_user(db: AsyncSession) -> User:
    return await make_user(db, "staff@test.com", "staff-pass-123", "staff")


@pytest.fixture
async def admin_user(db: AsyncSession) -> User:
    return await make_user(db, "admin@test.com", "admin-pass-123", "admin")


@pytest.fixture
def staff_headers(staff_user: User) -> dict:
    token = create_access_token(subject=staff_user.email, role=staff_user.role)
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def admin_headers(admin_user: User) -> dict:
    token = create_access_token(subject=admin_user.email, role=admin_user.role)
    return {"Authorization": f"Bearer {token}"}


# ---------- sample data ----------


SAMPLE_STUDENT = {
    "name": "Priya Sharma",
    "email": "priya@example.com",
    "phone": "+919876543210",
    "age": 17,
    "city": "Ahmedabad",
    "marks": 88,
}


@pytest.fixture
def sample_student() -> dict:
    return dict(SAMPLE_STUDENT)

Reading the changes

conn.run_sync(Base.metadata.create_all) create_all is a sync function. run_sync runs it inside the async connection. You will see this pattern whenever a sync-only SQLAlchemy helper has to be used with an async engine.

join_transaction_mode="create_savepoint" The async session is bound to a connection that already has a transaction open. This option tells SQLAlchemy: when the app code calls commit(), commit a savepoint inside the outer transaction instead of the outer transaction itself. The outer one is rolled back at the end, so each test is still isolated.

AsyncClient(transport=ASGITransport(app=app)) httpx's async client, talking to the app in-process through the ASGI interface. Same .get(), .post() API as TestClient, but every call is awaited and runs on the same event loop as the test and the session, which is what makes the dependency override work with asyncpg.

await db.flush() — same role as before, awaited.

The test files

Every test file changes by one mechanical rule:

Before	After
`def test_x(client, ...):`	`async def test_x(client, ...):`
`client.get(...)`, `client.post(...)` etc.	`await client.get(...)`, `await client.post(...)`
`crud.some_function(db, ...)`	`await crud.some_function(db, ...)`
`run_one(db)`	`await run_one(db)`
`db.refresh(x)`	`await db.refresh(x)`
`TestClient` type hints	`AsyncClient`
fixtures that call the client	`async def` with `await`

Here is test_auth.py fully converted so the pattern is concrete:

tests/test_auth.py

from httpx import AsyncClient


async def test_register_creates_staff_user(client: AsyncClient):
    response = await client.post("/auth/register", json={"email": "ritu@test.com", "password": "ritu-pass-123"})
    assert response.status_code == 201
    body = response.json()
    assert body["email"] == "ritu@test.com"
    assert body["role"] == "staff"
    assert body["is_active"] is True
    assert "password" not in body
    assert "hashed_password" not in body


async def test_register_duplicate_email_is_409(client: AsyncClient):
    payload = {"email": "ritu@test.com", "password": "ritu-pass-123"}
    assert (await client.post("/auth/register", json=payload)).status_code == 201
    assert (await client.post("/auth/register", json=payload)).status_code == 409


async def test_register_short_password_is_422(client: AsyncClient):
    response = await client.post("/auth/register", json={"email": "x@test.com", "password": "short"})
    assert response.status_code == 422


async def test_login_returns_token(client: AsyncClient, staff_user):
    response = await client.post("/auth/login", json={"email": "staff@test.com", "password": "staff-pass-123"})
    assert response.status_code == 200
    body = response.json()
    assert body["token_type"] == "bearer"
    assert len(body["access_token"]) > 20


async def test_login_wrong_password_is_401(client: AsyncClient, staff_user):
    response = await client.post("/auth/login", json={"email": "staff@test.com", "password": "wrong"})
    assert response.status_code == 401


async def test_login_unknown_email_is_401(client: AsyncClient):
    response = await client.post("/auth/login", json={"email": "nobody@test.com", "password": "whatever"})
    assert response.status_code == 401


async def test_me_without_token_is_401(client: AsyncClient):
    assert (await client.get("/auth/me")).status_code == 401


async def test_me_with_bad_token_is_401(client: AsyncClient):
    response = await client.get("/auth/me", headers={"Authorization": "Bearer not-a-real-token"})
    assert response.status_code == 401


async def test_me_returns_current_user(client: AsyncClient, staff_headers):
    response = await client.get("/auth/me", headers=staff_headers)
    assert response.status_code == 200
    assert response.json()["email"] == "staff@test.com"

And test_crud.py, where the structure changes the most:

tests/test_crud.py

import pytest
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from studydesk import crud
from studydesk.schemas import StudentCreate, UserCreate


async def make(db: AsyncSession, **overrides) -> int:
    data = {"name": "Priya Sharma", "email": "priya@example.com", "age": 17, "city": "Ahmedabad", "marks": 88}
    data.update(overrides)
    return (await crud.create_student(db, StudentCreate(**data))).id


async def test_create_and_get(db: AsyncSession):
    student_id = await make(db)
    student = await crud.get_student(db, student_id)
    assert student is not None
    assert student.name == "Priya Sharma"


async def test_get_by_email_is_case_insensitive(db: AsyncSession):
    await make(db)
    assert await crud.get_student_by_email(db, "PRIYA@EXAMPLE.COM") is not None


async def test_unique_email_enforced_by_database(db: AsyncSession):
    await make(db)
    with pytest.raises(IntegrityError):
        await make(db, name="Someone Else")
    await db.rollback()


async def test_list_filters_sorts_and_paginates(db: AsyncSession):
    await make(db, email="a@x.com", city="Surat", marks=72)
    await make(db, email="b@x.com", city="Surat", marks=45)
    await make(db, email="c@x.com", city="Ahmedabad", marks=95)

    page, total = await crud.list_students(db, city="surat", sort_by="marks", order="desc", limit=1)
    assert total == 2
    assert len(page) == 1
    assert page[0].marks == 72


async def test_update_and_delete(db: AsyncSession):
    student_id = await make(db)
    student = await crud.get_student(db, student_id)

    await crud.update_student(db, student, {"marks": 40})
    assert (await crud.get_student(db, student_id)).marks == 40

    await crud.delete_student(db, student)
    assert await crud.get_student(db, student_id) is None


async def test_authenticate_user(db: AsyncSession):
    await crud.create_user(db, UserCreate(email="Ritu@Test.com", password="ritu-pass-123"))
    assert await crud.authenticate_user(db, "ritu@test.com", "ritu-pass-123") is not None
    assert await crud.authenticate_user(db, "ritu@test.com", "wrong") is None
    assert await crud.authenticate_user(db, "nobody@test.com", "ritu-pass-123") is None


async def test_claim_next_job_takes_oldest_first(db: AsyncSession):
    first = await crud.create_job(db, kind="grade_report", payload={"n": 1})
    await crud.create_job(db, kind="grade_report", payload={"n": 2})

    claimed = await crud.claim_next_job(db)
    assert claimed.id == first.id
    assert claimed.status == "running"
    assert claimed.attempts == 1

test_students.py, test_jobs.py, and test_courses.py follow the same rule line by line (every def test_ → async def, every client. call and every crud./run_one/db.refresh call gets await; the five_students and three_students fixtures become async def and await their client calls; in test_jobs.py patch studydesk.tasks.asyncio.sleep with an async def that does nothing instead of time.sleep). The complete converted files are in the final repository walkthrough in Article 16.

Run:

uv run pytest

63 passed in 4.6s

Same sixty-three checks, now against the async stack.

Part 9: What Actually Improved

Re-run Neha's load test: 200 simultaneous GET /courses/1.

Before: 40 threads busy-waiting, 160 requests queued; the slowest responses took several seconds.
After: one event loop juggling 200 awaits; PostgreSQL's connection pool (default 5 connections, plus overflow) becomes the limit, and total time drops to a fraction.

The CPU is still mostly idle, but now that is because there is simply not much to compute, not because threads are stuck.

Where async bites

Three mistakes to watch for, because the errors are confusing:

Symptom	Cause	Fix
`MissingGreenlet: greenlet_spawn has not been called`	A relationship was lazy-loaded outside an `await` (often during response serialisation)	`selectinload` in the query, or `await db.refresh(obj, attribute_names=[...])`
Route returns `<coroutine object ...>` or does nothing	Called an `async def` without `await`	Add `await`
Whole API freezes during one request	A blocking call inside an `async def` route (`time.sleep`, `requests`, a sync DB driver)	Use the async equivalent, or run it in a thread with `await asyncio.to_thread(func, ...)`

Do you always need async?

No. The sync version of StudyDesk served the institute fine. Async matters when you have many concurrent requests that mostly wait: APIs that call other APIs, chat backends, anything with thousands of open connections. If your service does 20 requests per minute, the sync version is simpler to reason about, and that simplicity has value. Now you know both, and you know what the switch costs and buys.

Where StudyDesk is after this article

studydesk/
├── .env / docker-compose.yml       ← postgresql+asyncpg URLs
├── alembic/env.py                  ← swaps +asyncpg → +psycopg for migrations
├── pyproject.toml                  ← asyncpg; pytest-asyncio, asyncio_mode = auto
├── tests/                          ← async conftest + 63 async tests
└── src/studydesk/
    ├── database.py                 ← create_async_engine, AsyncSession, async get_db
    ├── crud.py                     ← all async; delete_student frees seats
    ├── dependencies.py             ← async get_current_user
    ├── routers/*.py                ← async def routes
    ├── tasks.py                    ← async generate_grade_report
    ├── worker.py                   ← asyncio loop
    └── seed.py                     ← asyncio.run

Concept Quiz

Answers are below. No coding needed.

Q1. What does await do, and how is it similar to yield in a generator?

Q2. What is the event loop?

Q3. Why is it not enough to change routes to async def while keeping the sync database driver?

Q4. What does expire_on_commit=False prevent in async code?

Q5. Why does send_welcome_email stay a synchronous function?

Q6. What does asyncio.run(loop()) do in the worker?

Q7. Why does Alembic use a +psycopg URL while the app uses +asyncpg?

Q8. What does join_transaction_mode="create_savepoint" do in the test db fixture?

Q9. Why must the test client be httpx.AsyncClient instead of TestClient now?

Q10. In which situations does async not help performance?

Answers

It pauses the current coroutine until the awaited operation finishes, letting the event loop run other coroutines meanwhile, then resumes at the same point. Like yield, it is a pause-and-resume point inside a function.
The scheduler that runs coroutines: it keeps track of which ones are paused on I/O and runs whichever is ready, on a single thread.
A blocking driver call inside an async def freezes the whole event loop for its duration, so every other request waits. Async only works if every wait is an await, all the way down to the driver.
Silent re-loading of attributes after a commit. In async, an implicit load outside an await raises MissingGreenlet; keeping values avoids it, and we refresh explicitly when needed.
BackgroundTasks runs plain functions on a thread pool, so its time.sleep does not block the loop. Making it async would gain nothing.
Creates an event loop, runs the loop() coroutine until it finishes (which, for the worker, is never), and closes the loop. It is the standard entry point for an async program.
Alembic's migration machinery is simplest with a sync connection. Swapping the driver name in env.py reuses the same .env value without a second variable.
When application code calls commit(), SQLAlchemy commits a savepoint inside the fixture's outer transaction instead of the outer transaction. The outer one is rolled back at the end, so the test still leaves no data behind.
AsyncClient with ASGITransport runs the app on the same event loop as the test and the async session. TestClient runs the app in a separate loop/thread, and asyncpg connections cannot be shared across loops.
CPU-bound work (a long calculation still blocks the loop), and low-traffic services where there is no waiting to overlap. Async pays off only when many requests spend most of their time waiting on I/O.

Epilogue

Two hundred requests, one event loop, and the slowest response is now measured in milliseconds. Neha nods at the graph.

"That's every topic on the list," she says. "Project setup, Pydantic, CRUD, CORS, PostgreSQL, Docker, auth, background jobs, migrations, tests, relationships, transactions, async. StudyDesk has all of it. Let's lay the whole thing out end to end, so anyone can read it from the top."

In the final article: StudyDesk complete — the full repository, every file in its final form, the one-command setup, and a map of which article built which piece.

Python - 20 Going Async: Async SQLAlchemy and Async Endpoints

Where StudyDesk is now

Part 1: The Idle CPU

Part 2: async and await, Explained

Blocking

Non-blocking

The event loop

The two rules

What async does not do

The one thing that must be true all the way down

Part 3: What Changes, in One Table

Update the URLs

Part 4: Async Engine and Session

Reading it

Alembic keeps a sync connection

Part 5: Async CRUD

Two things beyond the mechanical rule

Part 6: Async Dependencies and Routers

Part 7: Async Tasks, Worker, and Seed

Part 8: Async Tests

Setup

The new `conftest.py`

Reading the changes

The test files

Part 9: What Actually Improved

Where async bites

Do you always need async?

Where StudyDesk is after this article

Concept Quiz

Answers

Epilogue

Comments

More from this blog

Python 21 - StudyDesk Complete: The Final Project Walkthrough

Python 19 - Students and Courses: SQLAlchemy Relationships and Transactions

Python 18 - Proving It Works: pytest and API Testing

Python 17 - Changing the Database Safely: Alembic Migrations

Where StudyDesk is now

Part 1: The Idle CPU

Part 2: async and await, Explained

Blocking

Non-blocking

The event loop

The two rules

What async does not do

The one thing that must be true all the way down

Part 3: What Changes, in One Table

Update the URLs

Part 4: Async Engine and Session

Reading it

Alembic keeps a sync connection

Part 5: Async CRUD

Two things beyond the mechanical rule

Part 6: Async Dependencies and Routers

Part 7: Async Tasks, Worker, and Seed

Part 8: Async Tests

Setup

The new conftest.py

Reading the changes

The test files

Part 9: What Actually Improved

Where async bites

Do you always need async?

Where StudyDesk is after this article

Concept Quiz

Answers

Epilogue

Comments

python

More from this blog

Python 21 - StudyDesk Complete: The Final Project Walkthrough

Python 19 - Students and Courses: SQLAlchemy Relationships and Transactions

Python 18 - Proving It Works: pytest and API Testing

Python 17 - Changing the Database Safely: Alembic Migrations

The new `conftest.py`