# Python - 20 Going Async: Async SQLAlchemy and Async Endpoints

*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

```plaintext
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`."

![](https://cdn.hashnode.com/uploads/covers/68822b6ad516e7a593530cf9/204587a8-d025-4356-ac04-261fe84f934c.png align="center")

---

## Part 2: async and await, Explained

### Blocking

```python
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

```python
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

1. `await` **only works inside** `async def`**.**
2. **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:

```bash
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:

```plaintext
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:

```yaml
      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`

```python
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

```python
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`

```python
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`

```python
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`

```python
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`

```python
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`

```python
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`

```python
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:

```python
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`

```python
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`

```python
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`

```python
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:

```toml
[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`

```python
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`

```python
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`

```python
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:

```bash
uv run pytest
```

```plaintext
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.

![](https://cdn.hashnode.com/uploads/covers/68822b6ad516e7a593530cf9/b9004ed7-cfde-4151-9b48-3e6186779946.png align="center")

---

## Where StudyDesk is after this article

```plaintext
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

1. 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.
2. 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.
3. 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.
4. 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.
5. `BackgroundTasks` runs plain functions on a thread pool, so its `time.sleep` does not block the loop. Making it async would gain nothing.
6. 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.
7. 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.
8. 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.
9. `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.
10. 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.