Zhimalab

Zhimalab / Tutorials / FastAPI in 7 Days

DAY 1

Meet FastAPI

Start from zero and build your first API service with about 30 lines of code — and experience the magic of automatic docs.

🎯 Goal: Run your first API⏱️ Estimate: 45 min📦 Requires: Python 3.8+

1What is FastAPI?

FastAPI is a modern, high-performance Python web framework created by Sebastián Ramírez in 2018. It is built on Starlette (an async ASGI framework) and Pydantic (a data validation library), with native support for asynchronous programming.

⚡

Blazing performance

On par with Node.js and Go — one of the fastest Python frameworks.

📖

Automatic docs

Auto-generates interactive Swagger UI and ReDoc documentation out of the box.

🔐

Type safety

Leverages Python type hints for automatic request/response validation and conversion.

🚀

Fast development

Less boilerplate, friendly hints, and an excellent editor autocomplete experience.

FastAPI vs Flask vs Django? FastAPI is great for building API services and microservices; Flask is lightweight and flexible but synchronous-first; Django is full-stack but heavier. If you are building a pure API, FastAPI is the best choice today.

2Installation & first program

First, create a virtual environment and install FastAPI plus the ASGI server Uvicorn:

Terminal

# Create the project directory
mkdir myapi && cd myapi

# Create a virtual environment
python -m venv venv
source venv/bin/activate    # Windows: venv\Scripts\activate

# Install FastAPI and Uvicorn
pip install fastapi uvicorn[standard]

Create the main program file main.py:

main.py

from fastapi import FastAPI

app = FastAPI(title="My First API", version="1.0.0")

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!", "status": "running"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}

Start the development server:

Terminal

uvicorn main:app --reload

# What main:app means:
#   main -> the file name main.py
#   app  -> the instance variable app
# --reload enables hot reload; the server restarts when code changes

After startup you will see output similar to:

Output

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Application startup complete.

3Automatic docs — the killer feature

FastAPI automatically generates two sets of interactive API docs without writing a single extra line of code:

Docs URL	Name	Features
`http://localhost:8000/docs`	Swagger UI	Test every endpoint directly in the browser
`http://localhost:8000/redoc`	ReDoc	Elegant layout, great for API reference docs
`http://localhost:8000/openapi.json`	OpenAPI Schema	Machine-readable JSON specification

Try it yourself: open http://localhost:8000/docs, find GET /items/{item_id}, click "Try it out", enter item_id=42 and q=hello, then click Execute — you will see the JSON response immediately!

4Line-by-line walkthrough

main.py key lines

app = FastAPI(title="...")     # Create the app instance; set title, description, version
@app.get("/")                 # Route decorator: GET request, path "/"
def read_root():               # Synchronous handler (async def also works)
    return {"message": "..."}  # Return a dict; FastAPI converts it to JSON automatically
@app.get("/items/{item_id}")  # Path parameters are wrapped in {}
def read_item(item_id: int):   # Type annotation int -> automatic validation & conversion
    ...                        # Non-int values automatically return a 422 error

Sync vs async: functions declared with def run in a thread pool; functions declared with async def run on the event loop. If your function makes await calls (async database, HTTP requests), you must use async def.

Quiz

Check your understanding

In the command uvicorn main:app --reload, what does main:app mean?

A The app configuration file of the main module

B The app variable (FastAPI instance) in main.py

C A database named main and a table named app

D The app route prefix of the main application

✏️ Exercise

Add a /hello endpoint

Add a GET /hello endpoint in main.py that returns {"greeting": "Hello, world"}. When done, click to view the reference answer.

main.py

@app.get("/hello")
def hello():
    return {"greeting": "Hello, world"}

After saving, --reload restarts automatically; visit http://localhost:8000/hello to see the result.

🎯

Day 1 complete!

You have set up the FastAPI development environment and understood routing, automatic docs and the basic structure.

DAY 2

Routing & Parameters

Master path parameters, query parameters and automatic validation — the core charm of FastAPI's type system.

🎯 Goal: Master parameter passing⏱️ Estimate: 50 min

1Path parameters

Path parameters are declared with curly braces {}, and FastAPI validates and converts them automatically based on the type annotation:

main.py

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

# GET /users/42    -> {"user_id": 42} (converted to int automatically)
# GET /users/abc   -> 422 error: "value is not a valid integer"

A path can contain multiple parameters, in the same order they are declared:

main.py

@app.get("/users/{user_id}/posts/{post_id}")
async def get_post(user_id: int, post_id: int):
    return {"user_id": user_id, "post_id": post_id}

Route order matters! If you have both /users/me and /users/{user_id}, put /users/me first — otherwise me is passed in as user_id and triggers a type validation failure.

2Query parameters

Parameters declared outside the path that have a default value or a type annotation are treated as query parameters:

main.py

# A mock product list endpoint
items_db = ["Apple", "Banana", "Orange", "Grape", "Watermelon"]

@app.get("/items")
async def list_items(skip: int = 0, limit: int = 10):
    return items_db[skip : skip + limit]

# /items               -> skip=0, limit=10 (uses defaults)
# /items?skip=2        -> skips the first 2
# /items?skip=1&limit=3 -> skips 1, takes 3

Required vs optional query parameters: a parameter without a default is required (missing it causes an error); one with a default is optional. To make a parameter optional with no particular default, use Optional[str] = None.

3Parameter validation

Use Query to add validation constraints to query parameters:

main.py

from fastapi import Query

@app.get("/search")
async def search(
    q: str = Query(
        min_length=3,        # at least 3 characters
        max_length=50,       # at most 50 characters
        pattern="^[\w ]+$",  # regex: only letters, digits, underscores and spaces
        description="Search keyword"
    ),
    page: int = Query(default=1, ge=1),     # >= 1
    size: int = Query(default=10, ge=1, le=100)  # 1~100
):
    return {"q": q, "page": page, "size": size}

Validation	Keyword args	Applies to
Range	`gt, ge, lt, le`	int, float
Length	`min_length, max_length`	str
Regex	`pattern`	str
Path params	Use `Path()` instead of `Query()`	Path parameters

🔌 Interactive simulator: try parameter validation

Simulate requests to the /search endpoint and try different inputs to see the results

GET /search

q (keyword)

page

size

Waiting for request...

4Path parameters vs query parameters

Aspect	Path parameters	Query parameters
Declaration	`/items/{id}`	Function parameter with default value
URL location	In the path	`?key=value`
Required?	Required	Optional if it has a default
Semantics	Identify a resource	Filter / sort / paginate
Validator	`Path()`	`Query()`

Quiz

Check your understanding

Which URL correctly matches @app.get("/items/{item_id}") with item_id: int while also passing the query parameter q?

A /items?q=test/42

B /items/42?q=test

C /items/42&q=test

D /items?item_id=42&q=test

✏️ Exercise

Implement a paginated query endpoint

Implement GET /products supporting query parameters category (optional string), min_price (optional float ≥ 0), page (default 1, ≥ 1) and page_size (default 10, 1–50). Return these parameters as JSON.

main.py

from fastapi import Query
from typing import Optional

@app.get("/products")
async def list_products(
    category: Optional[str] = None,
    min_price: float = Query(default=None, ge=0),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=10, ge=1, le=50),
):
    return {
        "filters": {"category": category, "min_price": min_price},
        "pagination": {"page": page, "page_size": page_size},
    }

🎯

Day 2 complete!

You have mastered path parameters, query parameters and the parameter validation mechanism.

DAY 3

Request Bodies & Pydantic Models

Define data models with Pydantic to make request-body validation elegant and powerful.

🎯 Goal: Master data modeling⏱️ Estimate: 55 min

1Pydantic model basics

Pydantic is the data validation core of FastAPI. You define a class that inherits from BaseModel, declare fields and types, and Pydantic automatically handles validation, conversion and doc generation.

main.py

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(min_length=1, max_length=100, description="Product name")
    price: float = Field(gt=0, description="Price, must be greater than 0")
    is_offer: bool = False
    tags: list[str] = []  # empty list by default

# This model automatically appears in the request-body schema on /docs

2Receiving request bodies

Use the model as the type annotation of a function parameter and FastAPI knows to read JSON from the request body:

main.py

from fastapi import FastAPI

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    return {
        "received": item,
        "with_tax": item.price * 1.08,
    }

# Request body (POST http://localhost:8000/items/):
# {
#   "name": "Keyboard",
#   "price": 299.0,
#   "is_offer": true,
#   "tags": ["Electronics", "Peripherals"]
# }

How does FastAPI know if a parameter is a path parameter, query parameter or request body? The rule is simple: ① declared inside path {} → path parameter; ② a Pydantic model type → request body; ③ anything else → query parameter.

3Mixing parameter types

Path parameters, query parameters and request bodies can be used together in one function:

main.py

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,              # path parameter
    item: Item,                # request body
    q: str | None = None       # query parameter
):
    result = {"item_id": item_id, **item.model_dump()}
    if q:
        result["q"] = q
    return result

4Validation in practice

Pydantic automatically validates types and constraints, returning a structured 422 error when validation fails:

Validation failure example

// Request: POST /items/  body: {"name": "", "price": -5}
// Response 422:
{
  "detail": [
    {
      "type": "string_too_short",
      "loc": ["body", "name"],
      "msg": "String should have at least 1 character",
      "input": ""
    },
    {
      "type": "greater_than",
      "loc": ["body", "price"],
      "msg": "Input should be greater than 0",
      "input": -5
    }
  ]
}

5Nested models

Pydantic models can be nested to express complex data structures:

main.py

class Address(BaseModel):
    city: str
    street: str
    zip_code: str

class User(BaseModel):
    name: str
    age: int = Field(ge=0, le=150)
    address: Address        # nested model
    hobbies: list[str] = []

@app.post("/users")
async def create_user(user: User):
    return {"created": user}

# Example request body:
# {
#   "name": "Alice",
#   "age": 20,
#   "address": {"city": "Beijing", "street": "Zhongguancun", "zip_code": "100080"},
#   "hobbies": ["Coding", "Music"]
# }

model_dump() vs dict(): Pydantic v2 uses model_dump() to convert a model to a dict (.dict() from v1 is deprecated). model_json_schema() returns the JSON Schema.

Quiz

Check your understanding

In async def update_item(item_id: int, item: Item, q: str | None = None), what is item recognized as?

A A path parameter

B A query parameter

C A request body (because Item is a Pydantic model)

D A response header

✏️ Exercise

Design a course model

Create a Course model with: name (1–50 characters), credits (integer 1–10), teacher name, and an optional list of prerequisite courses (a nested Course list). Implement a POST /courses endpoint.

main.py

from pydantic import BaseModel, Field
from typing import Optional

class Course(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    credits: int = Field(ge=1, le=10)
    teacher: str
    prerequisites: list["Course"] = []  # self-referencing nesting

@app.post("/courses")
async def create_course(course: Course):
    return {"created": course}

🎯

Day 3 complete!

You have mastered Pydantic data modeling, request-body handling and nested models.

DAY 4

Response Models & Error Handling

Control output formats, manage status codes and handle errors gracefully — make your API more professional.

🎯 Goal: Take control of input & output⏱️ Estimate: 50 min

1response_model controls the output

By default FastAPI returns every field of the value returned by the function. Using response_model filters the output so only the fields you want are exposed — this matters for security:

main.py

class UserIn(BaseModel):
    username: str
    password: str       # the input includes the password
    email: str

class UserOut(BaseModel):  # the output has no password!
    username: str
    email: str

@app.post("/users", response_model=UserOut)
async def create_user(user: UserIn):
    return user  # receives UserIn but only returns UserOut fields
# Response: {"username": "...", "email": "..."}  <- no password!

Why do you need response_model? ① Security: filter sensitive fields (like password hashes); ② Docs: let /docs show an accurate response structure; ③ Consistency: the output format stays under your control no matter what the function returns internally.

2Status codes

Use the status_code parameter to set the HTTP status code for a successful response:

main.py

from fastapi import status

@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

# Common status codes:
# 200 OK            - request succeeded (GET/PUT default)
# 201 Created       - resource created (common for POST)
# 204 No Content    - success with no response body
# 400 Bad Request   - client request error
# 404 Not Found     - resource not found
# 422 Unprocessable - validation failed

3Handling errors with HTTPException

When business logic fails, raise HTTPException to return an error response with a status code and details:

main.py

from fastapi import HTTPException

items = {"foo": "The Foo Wrestlers"}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "ItemNotFound"}  # optional custom header
        )
    return {"item": items[item_id]}

# GET /items/bar -> 404
# {"detail": "Item not found"}

4Custom exception handlers

Define your own exception classes and register handlers to produce a unified error response format:

main.py

from fastapi import Request
from fastapi.responses import JSONResponse

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(UnicornException)
async def unicorn_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"code": 418, "message": f"Oops, {exc.name} broke something"},
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name)
    return {"name": name}

5Other response controls

main.py

from fastapi import Response

# Set response headers
@app.get("/custom-header")
async def read_header(response: Response):
    response.headers["X-Custom"] = "hello"
    return {"msg": "See the response headers"}

# Set a Cookie
from fastapi import response
@app.post("/login")
async def login(resp: Response):
    resp.set_cookie(key="token", value="abc123", httponly=True)
    return {"msg": "Logged in"}

Quiz

Check your understanding

Why should a create-user endpoint use response_model=UserOut instead of returning UserIn directly?

A Because UserIn is slow and UserOut is faster

B To filter sensitive fields (e.g. password) so they are not exposed to the client

C Because FastAPI does not allow returning a UserIn type

D To reduce JSON serialization time

✏️ Exercise

Implement a query endpoint with 404 handling

Use a dict to simulate a database users = {1:{"name":"Alice","age":20}, 2:{"name":"Bob","age":22}}. Implement GET /users/{uid}: return the user if they exist, otherwise raise a 404 HTTPException.

main.py

from fastapi import HTTPException

users = {1: {"name": "Alice", "age": 20}, 2: {"name": "Bob", "age": 22}}

@app.get("/users/{uid}")
async def get_user(uid: int):
    if uid not in users:
        raise HTTPException(status_code=404, detail=f"User {uid} not found")
    return users[uid]

🎯

Day 4 complete!

You have mastered response models, status codes and exception handling.

DAY 5

Dependency Injection & Middleware

FastAPI's dependency injection system is its soul — reuse logic, manage resources and control access, all through it.

🎯 Goal: Understand the DI mindset⏱️ Estimate: 55 min

1What is dependency injection?

Dependency Injection (DI) is a design pattern: instead of creating the objects you need yourself, you declare "what I need" and let the framework inject them. In FastAPI this is implemented with Depends().

In plain words: it is like ordering at a restaurant (declaring your needs) — the waiter brings the dish to you (injecting the dependency), and you do not worry about how it was cooked. FastAPI is the waiter, and Depends() is your order form.

2Your first dependency

main.py

from fastapi import Depends

# Define a dependency: just a plain function
def common_params(q: str | None = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

# Inject it into a route
@app.get("/items")
async def read_items(commons: dict = Depends(common_params)):
    return {"message": "Item list", "params": commons}

@app.get("/users")
async def read_users(commons: dict = Depends(common_params)):
    return {"message": "User list", "params": commons}
# Both endpoints share the same query-parameter logic without repetition!

3Nesting & caching dependencies

Dependencies can be nested (a dependency depending on another), and within the same request a dependency only runs once (the result is cached):

main.py

def query_db(q: str | None = None):
    if q:
        return {"query": q, "results": [f"result-{q}-1", f"result-{q}-2"]}
    return {"query": None, "results": []}

def logic_dep(db: dict = Depends(query_db)):
    # nested dependency: logic_dep depends on query_db
    return {"db": db, "extra": "extra logic"}

@app.get("/search")
async def search(
    dep1: dict = Depends(logic_dep),
    dep2: dict = Depends(query_db),  # query_db runs only once! dep2 reuses the cache
):
    return {"dep1": dep1, "dep2": dep2}

Disable caching with use_cache=False: if you need a dependency to re-run every time, use Depends(query_db, use_cache=False).

4Managing resources with dependencies (DB connections)

Dependencies are perfect for resources that must be opened/closed, combined with the yield syntax:

main.py

def get_db():
    db = "open database connection"  # mock
    try:
        yield db            # code before yield = runs before the request
        # the yielded value is injected into the route function
    finally:
        print("close database connection")  # after yield = runs after the request

@app.get("/data")
async def read_data(db = Depends(get_db)):
    return {"db_status": db}

5Global dependencies & router groups

main.py

def verify_token(x_token: str = Header()):
    if x_token != "secret-token":
        raise HTTPException(400, "Invalid X-Token")

# App-level global dependency: runs for every route
app = FastAPI(dependencies=[Depends(verify_token)])

# Router group (APIRouter) dependency
from fastapi import APIRouter
admin = APIRouter(prefix="/admin", dependencies=[Depends(verify_token)])

@admin.get("/dashboard")
async def dashboard():
    return {"area": "admin"}

app.include_router(admin)

6Middleware

Middleware runs before a request reaches a route and before the response returns to the client — ideal for cross-cutting concerns like logging, CORS and rate limiting:

main.py

import time
from fastapi import Request

@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)  # call the rest of the chain
    duration = time.time() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}s"
    return response

# CORS middleware (cross-origin)
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # frontend origin
    allow_methods=["*"],
    allow_headers=["*"],
)

Quiz

Check your understanding

Which statement about FastAPI's dependency caching is correct?

A Dependencies always re-run on every call and are never cached

B Within the same request, the same dependency runs once and the result is cached and reused

C The cache persists across requests, so the second request reuses the previous result

D Only async dependencies are cached; sync dependencies are not

✏️ Exercise

Reuse pagination parameters with a dependency

Create a pagination_dep dependency that accepts page (default 1, ≥ 1) and size (default 10, 1–100) and returns {"offset": (page-1)*size, "limit": size}. Use it in two endpoints.

main.py

from fastapi import Depends, Query

def pagination_dep(
    page: int = Query(default=1, ge=1),
    size: int = Query(default=10, ge=1, le=100),
):
    return {"offset": (page - 1) * size, "limit": size}

@app.get("/articles")
async def list_articles(pg: dict = Depends(pagination_dep)):
    return {"type": "articles", "pagination": pg}

@app.get("/comments")
async def list_comments(pg: dict = Depends(pagination_dep)):
    return {"type": "comments", "pagination": pg}

🎯

Day 5 complete!

You understand dependency injection — reusing logic, managing resources and configuring middleware.

DAY 6

Database Integration & Async

Connect a database with SQLAlchemy, implement full CRUD and feel the power of asynchronous programming.

🎯 Goal: Database CRUD⏱️ Estimate: 60 min

1Project structure

As the project grows, split the code across multiple files. A recommended directory structure:

Project structure

myapi/
├── main.py              # app entry point
├── database.py          # database connection
├── models.py            # SQLAlchemy models
├── schemas.py           # Pydantic models (input/output)
├── crud.py              # database operations
└── requirements.txt

2Database connection layer

database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DB_URL = "sqlite:///./app.db"  # SQLite file database

engine = create_engine(
    SQLALCHEMY_DB_URL,
    connect_args={"check_same_thread": False}  # SQLite-specific
)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base = declarative_base()

# Dependency: each request gets its own session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3SQLAlchemy models vs Pydantic models

Aspect	SQLAlchemy model	Pydantic model
Purpose	Maps database tables	Validates request/response data
Definition	Inherit Base, use Column	Inherit BaseModel
File	models.py	schemas.py
Focus	How data is stored	How data is transferred

models.py

from sqlalchemy import Column, Integer, String, Float
from database import Base

class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False)
    price = Column(Float, nullable=False)
    description = Column(String(500), default="")

schemas.py

from pydantic import BaseModel, Field

class ProductCreate(BaseModel):  # used when creating
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)
    description: str = ""

class ProductOut(BaseModel):     # used when returning (includes id)
    id: int
    name: str
    price: float
    description: str

    class Config:
        from_attributes = True   # allow reading attributes from ORM objects

4CRUD operations

crud.py

from sqlalchemy.orm import Session
import models, schemas

def get_product(db: Session, product_id: int):
    return db.query(models.Product).filter(
        models.Product.id == product_id
    ).first()

def get_products(db: Session, skip: int = 0, limit: int = 20):
    return db.query(models.Product).offset(skip).limit(limit).all()

def create_product(db: Session, product: schemas.ProductCreate):
    db_product = models.Product(**product.model_dump())
    db.add(db_product)
    db.commit()
    db.refresh(db_product)  # get the auto-increment id
    return db_product

def delete_product(db: Session, product_id: int):
    product = get_product(db, product_id)
    if product:
        db.delete(product)
        db.commit()
    return product

5Wiring up the routes

main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
import models, schemas, crud
from database import engine, get_db

models.Base.metadata.create_all(bind=engine)  # create tables
app = FastAPI()

@app.post("/products", response_model=schemas.ProductOut, status_code=201)
def create_product(product: schemas.ProductCreate, db: Session = Depends(get_db)):
    return crud.create_product(db, product)

@app.get("/products", response_model=list[schemas.ProductOut])
def list_products(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
    return crud.get_products(db, skip, limit)

@app.get("/products/{pid}", response_model=schemas.ProductOut)
def read_product(pid: int, db: Session = Depends(get_db)):
    product = crud.get_product(db, pid)
    if not product:
        raise HTTPException(404, "Product not found")
    return product

@app.delete("/products/{pid}")
def remove_product(pid: int, db: Session = Depends(get_db)):
    product = crud.delete_product(db, pid)
    if not product:
        raise HTTPException(404, "Product not found")
    return {"deleted": pid}

Sync vs async databases: the code above uses synchronous SQLAlchemy. For async, you need aiosqlite/asyncpg + AsyncSession, functions written with async def and await db.execute(). Beginners are advised to master the sync version first.

6Async programming basics

FastAPI natively supports async/await. Async matters in I/O-bound scenarios (databases, HTTP requests, file I/O) because it does not block threads:

main.py

import httpx  # async HTTP client
import asyncio

@app.get("/weather/{city}")
async def get_weather(city: str):
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://wttr.in/{city}", params={"format": "j1"}
        )
        data = resp.json()
    return {"city": city, "temp": data["current_condition"][0]["temp_C"]}

# Request multiple external APIs at once (concurrency)
@app.get("/multi")
async def multi():
    async with httpx.AsyncClient() as client:
        tasks = [client.get(f"https://httpbin.org/delay/{i}") for i in range(3)]
        responses = await asyncio.gather(*tasks)  # run concurrently!
    return {"count": len(responses)}

When should you use async def? When your function makes await calls (async I/O), you must use async def. For pure computation or synchronous I/O, a plain def is fine — FastAPI runs it in a thread pool without blocking the event loop.

Quiz

Check your understanding

In a FastAPI + SQLAlchemy project, why is the database session injected via Depends(get_db) instead of being created inside the function?

A Because Depends performs better than creating it directly

B Because FastAPI requires the use of Depends

C To ensure each request uses its own session that closes automatically when the request ends

D So that /docs can display database connection information

✏️ Exercise

Implement an update endpoint

Implement PUT /products/{pid} in crud.py and main.py. Create a ProductUpdate schema (all fields optional), update the non-empty fields and return the result.

crud.py + main.py

# schemas.py
from typing import Optional
class ProductUpdate(BaseModel):
    name: Optional[str] = None
    price: Optional[float] = None
    description: Optional[str] = None

# crud.py
def update_product(db: Session, pid: int, data: schemas.ProductUpdate):
    product = get_product(db, pid)
    if not product:
        return None
    for key, val in data.model_dump(exclude_unset=True).items():
        setattr(product, key, val)
    db.commit()
    db.refresh(product)
    return product

# main.py
@app.put("/products/{pid}", response_model=schemas.ProductOut)
def update(pid: int, data: schemas.ProductUpdate, db: Session = Depends(get_db)):
    product = crud.update_product(db, pid, data)
    if not product:
        raise HTTPException(404, "Product not found")
    return product

🎯

Day 6 complete!

You can integrate a database, implement full CRUD and understand asynchronous programming.

DAY 7

Auth, Testing & Deployment

JWT authentication, automated testing and production deployment — take your API to the real world.

🎯 Goal: Ready for production⏱️ Estimate: 65 min

1OAuth2 + JWT authentication

JWT (JSON Web Token) is a common approach for API authentication: after logging in, the user receives a token and sends it with subsequent requests to prove their identity.

auth.py

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Password hashing
def hash_password(password: str) -> str:
    return pwd_ctx.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_ctx.verify(plain, hashed)

# Generate a token
def create_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

# Current-user dependency
def get_current_user(token: str = Depends(oauth2_scheme)):
    cred_err = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise cred_err
    except JWTError:
        raise cred_err
    # In practice you would look the user up in the database; simplified here
    return {"username": username}

The login endpoint issues a token, and protected endpoints use Depends(get_current_user):

main.py

from fastapi.security import OAuth2PasswordRequestForm
from auth import create_token, get_current_user, verify_password, hash_password

# Mock user store (use a real database in practice)
fake_users = {"alice": {"username": "alice", "hashed": hash_password("secret")}}

@app.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends()):
    user = fake_users.get(form.username)
    if not user or not verify_password(form.password, user["hashed"]):
        raise HTTPException(401, "Incorrect username or password")
    token = create_token({"sub": user["username"]})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/me")
async def me(current = Depends(get_current_user)):
    return current

2Automated testing

FastAPI provides TestClient, built on httpx, so you can test without actually starting the server:

test_main.py

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, FastAPI!", "status": "running"}

def test_read_item():
    response = client.get("/items/42", params={"q": "test"})
    assert response.status_code == 200
    assert response.json()["item_id"] == 42
    assert response.json()["q"] == "test"

def test_invalid_item_id():
    response = client.get("/items/not-a-number")
    assert response.status_code == 422  # validation failed

def test_create_item():
    response = client.post("/items/", json={
        "name": "Keyboard", "price": 299.0, "is_offer": True, "tags": ["Peripherals"]
    })
    assert response.status_code == 200
    assert response.json()["received"]["name"] == "Keyboard"

Run tests

pip install pytest
pytest test_main.py -v

# Example output:
# test_main.py::test_read_root PASSED
# test_main.py::test_read_item PASSED
# test_main.py::test_invalid_item_id PASSED
# test_main.py::test_create_item PASSED

3Environment variables & config

config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My API"
    database_url: str = "sqlite:///./app.db"
    secret_key: str = "dev-secret"
    debug: bool = True

    class Config:
        env_file = ".env"

settings = Settings()
# Reads .env files or environment variables, type-safe

.env

APP_NAME=Production API
DATABASE_URL=postgresql://user:pass@localhost:5432/mydb
SECRET_KEY=a-very-long-random-string
DEBUG=False

4Deploy to production

Deploy commands

# Run Uvicorn workers with Gunicorn (recommended for production)
pip install gunicorn

gunicorn main:app \
  -w 4 \                    # 4 worker processes
  -k uvicorn.workers.UvicornWorker \
  -b 0.0.0.0:8000

# Docker deployment
# Dockerfile:
#   FROM python:3.12-slim
#   WORKDIR /app
#   COPY requirements.txt .
#   RUN pip install --no-cache-dir -r requirements.txt
#   COPY . .
#   CMD ["gunicorn", "main:app", "-w", "4", \
#        "-k", "uvicorn.workers.UvicornWorker", \
#        "-b", "0.0.0.0:8000"]

Scenario	Command	Notes
Development	`uvicorn main:app --reload`	Hot reload, single process
Production	`gunicorn ... -k UvicornWorker`	Multiple workers, stable
Docker	`docker build -t myapi . && docker run -p 8000:8000 myapi`	Containerized deployment
Reverse proxy	Nginx → Uvicorn	Handles TLS, static files, load balancing

5Advanced roadmap

🔌

WebSocket

Real-time bidirectional communication, great for chat and push.

📄

Background tasks

BackgroundTasks for emails, Celery for heavy work.

🔍

APIRouter

Split routes for large projects and microservices.

📊

Pagination / caching

Redis caching, cursor pagination.

🐳

Docker Compose

Orchestrate API + DB + redis containers.

📈

Monitoring

Prometheus + Grafana metrics.

🎉

Congratulations on finishing the 7-day crash course!

From Hello World to auth, testing and deployment — you now have the fundamentals to build production-grade APIs with FastAPI. The best way to keep learning is: build a project.

Quiz

Check your understanding

Which statement about JWT token authentication is correct?

A The token is stored on the server and every request is verified against the server

B The token contains a signature; the server verifies it with a secret key without querying the database

C A token never expires once issued

D JWT can only be sent over HTTPS and does not work over HTTP

✏️ Exercise

Write a test case

Write a test for the POST /items/ endpoint that verifies a 422 status code is returned when price is negative.

test_main.py

def test_create_item_invalid_price():
    response = client.post("/items/", json={
        "name": "Test product", "price": -10.0
    })
    assert response.status_code == 422
    # you can inspect the error details further
    detail = response.json()["detail"]
    assert any(d["loc"] == ["body", "price"] for d in detail)

🏆

All lessons complete!

You have completed the FastAPI crash course. Mark it done and record your achievement!