Advanced 16 min

Async, Databases and Deployment

When to use async def, talking to a database and running in production.

What you will learn

Choose async vs sync
Connect SQLAlchemy
Run with uvicorn/Docker

You may write endpoints with def or async def. Choosing correctly affects performance, and databases plus deployment are what turn a demo into a service.

async or not?

async def: use when everything you call inside is awaitable (async database drivers, httpx.AsyncClient). The event loop can serve other requests while you wait.
def: use when you call blocking code (a regular database driver, requests, heavy computation). FastAPI runs these in a thread pool so they do not freeze the server.

The trap

Calling blocking code inside async def stalls the whole event loop and every other request waits. When unsure, use plain def.

import httpx

@app.get("/proxy")
async def proxy():
    async with httpx.AsyncClient() as client:
        r = await client.get("https://api.github.com")
    return {"status": r.status_code}

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

engine = create_engine("sqlite:///./app.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False)

class Base(DeclarativeBase):
    pass

# models.py
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

class Todo(Base):
    __tablename__ = "todos"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    done: Mapped[bool] = mapped_column(default=False)

Base.metadata.create_all(engine)

@app.post("/todos", status_code=201)
def create(t: TodoIn, db: Session = Depends(get_db)):
    todo = Todo(title=t.title)
    db.add(todo)
    db.commit()
    db.refresh(todo)
    return todo

@app.get("/todos/{tid}")
def read(tid: int, db: Session = Depends(get_db)):
    todo = db.get(Todo, tid)
    if not todo:
        raise HTTPException(404)
    return todo

For production use PostgreSQL, and manage schema changes with Alembic migrations instead of create_all.

from fastapi import BackgroundTasks

def send_email(to: str):
    print("emailing", to)

@app.post("/signup")
def signup(email: str, bg: BackgroundTasks):
    bg.add_task(send_email, email)      # runs after the response is sent
    return {"ok": True}

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Read secrets and the database URL from environment variables (pydantic-settings helps).
Run behind a process manager or several workers: uvicorn main:app --workers 4.
Add CORS middleware if a browser front end on another origin calls your API.
Write tests with TestClient from fastapi.testclient.

Try it yourself

Write a test using TestClient that posts a todo and asserts the response has status 201 and a title field.

Show solution

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_todo():
    r = client.post("/todos", json={"title": "write tests"})
    assert r.status_code == 201
    assert r.json()["title"] == "write tests"