FastAPI Web 开发从入门到实战（详细版）

2026年10月9日约 5 分钟

对照官方文档风格的中文长文：与 Flask/Django 对比、uvicorn 安装、首个应用、路径/查询/请求体、Pydantic、依赖注入、Router、CORS、JWT/OAuth2PasswordBearer、SQLAlchemy+SQLite CRUD、OpenAPI、TestClient、免费档部署与常见坑。

FastAPI 是基于 Starlette（ASGI）与 Pydantic 的现代 Python Web 框架：用类型注解描述入参与响应，自动生成 OpenAPI（Swagger UI / ReDoc），同步与异步路由都能写。本篇按官方文档的思路组织，尽量给出可直接粘贴改跑的片段；默认假设 Python 3.10+。

和 Flask / Django 比一句：Flask 是同步 WSGI 微内核，模板与表单生态成熟，自由度高；Django 电池齐全（Admin、ORM、用户系统），适合约定统一的大型单体；FastAPI 以类型注解 + OpenAPI 见长，JSON API、异步 I/O、自动文档与 schema 校验最省心。选库看团队约定与需求形状，不是谁更「高级」。

1. 简介：何时选 FastAPI

适合：

前后端分离的 REST / JSON API
需要开箱的交互式文档（/docs、/redoc）
想用类型注解驱动校验与编辑器补全
异步数据库、外部 HTTP、WebSocket 等 I/O 密集场景

不太适合或要慎重：

以服务端渲染 HTML 为主、强依赖 Jinja 表单生态 → Flask / Django 往往更顺手
需要开箱 Admin + 完整用户权限后台 → Django 通常更省心
纯 CPU 密集计算 → 异步帮不上忙，仍要进程/线程模型

对多数「先把 API 跑通再演进」的项目，先把路由、Pydantic 模型、依赖注入和项目分层想清楚，再谈 ORM 与鉴权。

2. 环境：venv、pip、uvicorn

2.1 虚拟环境与安装

python3 -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
python -m pip install -U pip
pip install "fastapi[standard]"
# 等价于：fastapi + uvicorn[standard] + 常用开发依赖
# 若只要核心：pip install fastapi uvicorn

验证：

python -c "import fastapi, uvicorn; print(fastapi.__version__, uvicorn.__version__)"

建议钉死依赖：

pip freeze > requirements.txt
# 之后：pip install -r requirements.txt

2.2 运行服务器：uvicorn

FastAPI 应用是 ASGI app，开发时用 uvicorn 热重载：

uvicorn main:app --reload --host 127.0.0.1 --port 8000
# 浏览器：http://127.0.0.1:8000/docs

main:app 表示模块 main.py 里的变量 app。生产关掉 --reload，并按需加 --workers（多进程时注意共享状态与数据库连接）。

3. 第一个应用

main.py：

from fastapi import FastAPI

app = FastAPI(
    title="Notes API",
    description="FastAPI 入门示例",
    version="0.1.0",
)

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

@app.get("/health")
async def health():
    return {"status": "ok"}

uvicorn main:app --reload
curl http://127.0.0.1:8000/
# {"message":"Hello, FastAPI!"}

同步 def 与异步 async def 都可以：纯 CPU / 同步库用 def（跑在线程池）；await 异步 I/O 用 async def。不要在 async def 里直接调用阻塞的同步 ORM / requests，否则会卡住事件循环。

打开 http://127.0.0.1:8000/docs 即可看到自动生成的 Swagger UI。

4. 路径参数、查询参数、请求体

4.1 路径参数（Path）

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(
    item_id: int = Path(..., ge=1, description="物品 ID，从 1 开始"),
):
    return {"item_id": item_id}

类型注解会做校验：/items/abc → 422。Path(...) 的 ... 表示必填。

4.2 查询参数（Query）

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/search")
def search(
    q: Annotated[str | None, Query(min_length=1, max_length=50)] = None,
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
):
    return {"q": q, "skip": skip, "limit": limit}

/search?q=fastapi&skip=0&limit=10。有默认值的是可选查询参数。

4.3 请求体（Body）与混合用法

from pydantic import BaseModel, Field
from fastapi import FastAPI

app = FastAPI()

class ItemIn(BaseModel):
    name: str = Field(min_length=1, max_length=80)
    price: float = Field(gt=0)
    tags: list[str] = []

@app.post("/items/")
def create_item(item: ItemIn):
    return {"ok": True, "item": item}

@app.put("/items/{item_id}")
def update_item(item_id: int, item: ItemIn, notify: bool = False):
    # 路径 + 体 + 查询可同时出现；单个 BaseModel 默认当 JSON body
    return {"item_id": item_id, "item": item, "notify": notify}

curl -X POST http://127.0.0.1:8000/items/ \
  -H 'Content-Type: application/json' \
  -d '{"name":"笔记本","price":12.5,"tags":["文具"]}'

多个 body 模型时用 Body(embed=True) 等技巧，见官方 Body - Multiple Parameters。

5. Pydantic 模型：请求 / 响应分离

生产中建议 In / Out / DB 分层，避免把密码哈希、内部字段泄露给客户端：

from datetime import datetime
from pydantic import BaseModel, ConfigDict, EmailStr, Field

class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8, max_length=128)
    display_name: str = Field(min_length=1, max_length=40)

class UserPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # ORM 对象可直接校验

    id: int
    email: EmailStr
    display_name: str
    created_at: datetime

class UserInDB(UserPublic):
    hashed_password: str

from fastapi import FastAPI

app = FastAPI()

@app.post("/users/", response_model=UserPublic, status_code=201)
def register(user: UserCreate):
    # 伪代码：hash 密码后入库，返回不含 password 的 UserPublic
    now = datetime.utcnow()
    return UserPublic(
        id=1,
        email=user.email,
        display_name=user.display_name,
        created_at=now,
    )

response_model 会按模型过滤输出；EmailStr 需要 pip install email-validator（fastapi[standard] 通常已带）。

更多：Pydantic Models、Response Model。

6. 依赖注入（Depends）

依赖注入是 FastAPI 的核心模式：鉴权、数据库会话、分页、公共校验都可抽成可复用函数。

from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()

def get_pagination(skip: int = 0, limit: int = 20):
    return {"skip": max(skip, 0), "limit": min(limit, 100)}

Pagination = Annotated[dict, Depends(get_pagination)]

async def get_token_header(x_token: Annotated[str | None, Header()] = None):
    if x_token != "secret-demo-token":
        raise HTTPException(status_code=401, detail="Invalid X-Token")
    return x_token

@app.get("/notes/")
def list_notes(page: Pagination):
    return {"items": [], **page}

@app.get("/private")
async def private(token: Annotated[str, Depends(get_token_header)]):
    return {"ok": True, "token": token}

子依赖可以嵌套；yield 依赖适合「用完关闭」的资源（如 DB session）：

from collections.abc import Generator

def get_db() -> Generator:
    db = "假会话"  # 换成 SessionLocal()
    try:
        yield db
    finally:
        pass  # db.close()

官方：Dependencies。

7. APIRouter：模块化路由

app/
  __init__.py
  main.py          # FastAPI() + include_router
  deps.py          # 公共 Depends
  routers/
    notes.py
    auth.py
  schemas/
    note.py
  models/
    note.py

app/routers/notes.py：

from fastapi import APIRouter, HTTPException

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

_FAKE_DB: dict[int, dict] = {}

@router.get("/")
def list_notes():
    return list(_FAKE_DB.values())

@router.get("/{note_id}")
def get_note(note_id: int):
    note = _FAKE_DB.get(note_id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")
    return note

@router.post("/", status_code=201)
def create_note(title: str, body: str = ""):
    note_id = len(_FAKE_DB) + 1
    note = {"id": note_id, "title": title, "body": body}
    _FAKE_DB[note_id] = note
    return note

app/main.py：

from fastapi import FastAPI
from app.routers import notes, auth

app = FastAPI(title="Notes API")
app.include_router(notes.router)
app.include_router(auth.router)

uvicorn app.main:app --reload

tags 会出现在 /docs 分组里。官方：Bigger Applications - Multiple Files。

8. CORS：前后端分离必配

浏览器跨域时需显式允许源。开发期可放宽，生产务必收紧。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost:5173",   # Vite 前端
    "http://127.0.0.1:5173",
    # "https://your-frontend.example.com",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,          # 不要用 ["*"] 再配 allow_credentials=True
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

带 Cookie 的跨站还要考虑 SameSite 与 HTTPS。官方：CORS。

9. 简单鉴权：OAuth2PasswordBearer + JWT

下面是教学用最小实现：密码用哈希存放，登录换 JWT，受保护路由用 Depends 取当前用户。生产请换强密钥、加 refresh、设合理过期与吊销策略。

pip install "python-jose[cryptography]" passlib bcrypt
# 若 bcrypt 版本冲突，可改用 pwdlib / 按官方 Security 章节更新

app/security.py：

from datetime import datetime, timedelta, timezone
from typing import Annotated

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

SECRET_KEY = "change-me-in-production-use-openssl-rand-hex-32"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60

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

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    username: str | None = None

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

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(subject: str, expires_delta: timedelta | None = None) -> str:
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return jwt.encode(
        {"sub": subject, "exp": expire},
        SECRET_KEY,
        algorithm=ALGORITHM,
    )

# 演示用内存用户；实战换成数据库
fake_users_db = {
    "alice": {
        "username": "alice",
        "hashed_password": hash_password("secret123"),
        "disabled": False,
    }
}

def authenticate_user(username: str, password: str):
    user = fake_users_db.get(username)
    if not user or not verify_password(password, user["hashed_password"]):
        return None
    return user

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    credentials_exc = 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: str | None = payload.get("sub")
        if username is None:
            raise credentials_exc
    except JWTError as exc:
        raise credentials_exc from exc
    user = fake_users_db.get(username)
    if user is None or user.get("disabled"):
        raise credentials_exc
    return user

app/routers/auth.py：

from datetime import timedelta
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm

from app.security import (
    Token,
    ACCESS_TOKEN_EXPIRE_MINUTES,
    authenticate_user,
    create_access_token,
    get_current_user,
)

router = APIRouter(tags=["auth"])

@router.post("/token", response_model=Token)
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token = create_access_token(
        subject=user["username"],
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return Token(access_token=token)

@router.get("/users/me")
async def read_me(current_user: Annotated[dict, Depends(get_current_user)]):
    return {
        "username": current_user["username"],
        "disabled": current_user["disabled"],
    }

OAuth2PasswordRequestForm 期望表单字段 username / password（不是 JSON），Swagger 的 Authorize 按钮会走同一套。

curl -X POST http://127.0.0.1:8000/token \
  -d 'username=alice&password=secret123'
# {"access_token":"...","token_type":"bearer"}

curl http://127.0.0.1:8000/users/me \
  -H "Authorization: Bearer <上一步的 token>"

官方：Security、OAuth2 with Password (and hashing), Bearer with JWT。
注意：示例密钥不可用于生产；JWT 只是声明的编码，不是加密传输——务必 HTTPS。

10. SQLAlchemy + SQLite：同步 CRUD（可改为异步）

10.1 同步版本（好上手，兼容面广）

pip install sqlalchemy

app/database.py：

from collections.abc import Generator

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker, Session

SQLALCHEMY_DATABASE_URL = "sqlite:///./notes.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},  # SQLite + 多线程需要
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Base(DeclarativeBase):
    pass

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app/models/note.py：

from datetime import datetime

from sqlalchemy import DateTime, Integer, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base

class Note(Base):
    __tablename__ = "notes"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(120), nullable=False)
    body: Mapped[str] = mapped_column(Text, default="")
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

app/schemas/note.py：

from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field

class NoteCreate(BaseModel):
    title: str = Field(min_length=1, max_length=120)
    body: str = ""

class NoteUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=120)
    body: str | None = None

class NoteOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    body: str
    created_at: datetime | None = None

app/routers/notes_db.py：

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

from app.database import get_db
from app.models.note import Note
from app.schemas.note import NoteCreate, NoteOut, NoteUpdate

router = APIRouter(prefix="/notes", tags=["notes"])
DbDep = Annotated[Session, Depends(get_db)]

@router.get("/", response_model=list[NoteOut])
def list_notes(db: DbDep, skip: int = 0, limit: int = 50):
    return db.query(Note).offset(skip).limit(min(limit, 100)).all()

@router.post("/", response_model=NoteOut, status_code=status.HTTP_201_CREATED)
def create_note(payload: NoteCreate, db: DbDep):
    note = Note(title=payload.title, body=payload.body)
    db.add(note)
    db.commit()
    db.refresh(note)
    return note

@router.get("/{note_id}", response_model=NoteOut)
def get_note(note_id: int, db: DbDep):
    note = db.get(Note, note_id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")
    return note

@router.patch("/{note_id}", response_model=NoteOut)
def update_note(note_id: int, payload: NoteUpdate, db: DbDep):
    note = db.get(Note, note_id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")
    data = payload.model_dump(exclude_unset=True)
    for k, v in data.items():
        setattr(note, k, v)
    db.commit()
    db.refresh(note)
    return note

@router.delete("/{note_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_note(note_id: int, db: DbDep):
    note = db.get(Note, note_id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")
    db.delete(note)
    db.commit()
    return None

启动时建表（小项目可接受；正式项目用 Alembic 迁移）：

# app/main.py 片段
from app.database import Base, engine
from app.routers import notes_db

Base.metadata.create_all(bind=engine)
app.include_router(notes_db.router)

10.2 异步版本要点（SQLAlchemy 2 + aiosqlite）

pip install sqlalchemy aiosqlite

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

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

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

# 路由里：
# result = await db.execute(select(Note))
# notes = result.scalars().all()

异步路由必须 async def，并用 select() / await db.execute(...)，不要混用同步 db.query()。官方与 SQLAlchemy 文档：Async SQLAlchemy。

SQLite 适合本机与单实例 demo；多实例或高并发写请换 PostgreSQL 等，并注意连接池与迁移。

11. OpenAPI 文档：/docs 与 /redoc

FastAPI 默认：

路径	说明
`/docs`	Swagger UI，可在线试请求
`/redoc`	ReDoc 阅读型文档
`/openapi.json`	OpenAPI schema

可在构造时定制：

app = FastAPI(
    title="Notes API",
    summary="个人笔记 API",
    version="1.2.0",
    contact={"name": "API Support", "email": "dev@example.com"},
    license_info={"name": "MIT"},
    # docs_url=None,  # 生产可关闭交互文档
    # redoc_url=None,
    # openapi_url="/api/v1/openapi.json",
)

路由上的 summary / description / response_model / status_code / tags / responses 都会进文档。鉴权方案（如 OAuth2PasswordBearer）会在 UI 显示 Authorize。

官方：Metadata and Docs URLs。

12. 测试：TestClient / httpx

pip install httpx pytest

tests/conftest.py：

import pytest
from fastapi.testclient import TestClient

from app.main import app

@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

tests/test_notes.py：

def test_health(client):
    r = client.get("/health")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"

def test_create_and_list_note(client):
    r = client.post("/notes/", json={"title": "hello", "body": "world"})
    assert r.status_code == 201
    note = r.json()
    assert note["title"] == "hello"

    r2 = client.get("/notes/")
    assert r2.status_code == 200
    assert any(n["id"] == note["id"] for n in r2.json())

def test_login_and_me(client):
    r = client.post(
        "/token",
        data={"username": "alice", "password": "secret123"},
    )
    assert r.status_code == 200
    token = r.json()["access_token"]
    r2 = client.get(
        "/users/me",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert r2.status_code == 200
    assert r2.json()["username"] == "alice"

pytest -q

TestClient 基于 httpx，可同步测 async def 路由。异步测试也可用 httpx.AsyncClient + ASGITransport。官方：Testing。

测数据库时建议：独立测试库或内存 SQLite，fixture 里 create_all / drop_all，并 dependency_overrides[get_db] = ... 覆盖依赖。

13. 部署到免费档：思路简述（不绑付费）

原则：用 uvicorn / gunicorn+uvicorn worker 跑 ASGI，前面用反向代理或平台边缘提供 HTTPS；SQLite 在多实例或无持久盘时不可靠。

13.1 本机 / VPS：uvicorn + Cloudflare Tunnel

pip install uvicorn
uvicorn app.main:app --host 127.0.0.1 --port 8000
# 生产可：
# gunicorn -k uvicorn.workers.UvicornWorker -w 2 -b 127.0.0.1:8000 app.main:app

Cloudflare Tunnel（cloudflared）可把内网端口映射到域名并带 HTTPS，适合演示与个人小工具。密钥与 SECRET_KEY 走环境变量，进程用 systemd 守护。

13.2 Render / Fly.io / Railway 简述

Render：Web Service，启动命令例如 uvicorn app.main:app --host 0.0.0.0 --port $PORT；免费档可能休眠。
Fly.io：容器部署，同样暴露 ASGI；注意机器休眠与卷。
Railway 等：大同小异，看平台文档设 PORT 与环境变量。

Dockerfile 示意：

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

无论哪家：关 debug/热重载、强密钥、生产可关 /docs、数据库按平台限制选型（免费 Postgres 优于多实例 SQLite）。本文不绑定任何付费方案。

官方：Deployment。

14. 常见坑与官方文档

常见坑

在 async def 里调用阻塞同步库（requests、同步 SQLAlchemy session）→ 事件循环卡住；改线程池或异步驱动。
生产仍用 --reload / 弱 SECRET_KEY → 性能与安全风险。
allow_origins=["*"] 且 allow_credentials=True → 浏览器会拒绝；CORS 配错难查。
把 ORM 模型直接当响应 → 泄露字段；用 response_model / Out schema。
SQLite 路径与多 worker → 工作目录变化「找不到库」；多进程写易锁。
忘记 Depends(get_db) 的 yield 关闭 → 连接泄漏。
JWT 当「加密」 → JWT 默认可解码，只做签名校验；传输靠 HTTPS。
OAuth2PasswordRequestForm 用 JSON 调 → 应为 form-urlencoded。
422 一脸懵 → 看响应体 detail；多半是类型/缺字段/校验失败。
依赖没进 requirements.txt → 线上 ImportError。
全局可变状态当缓存 → 多 worker 不共享；用 Redis 等。
测试未覆盖依赖 → 用 app.dependency_overrides 注入假 DB / 假用户。

官方与延伸链接

FastAPI 文档：https://fastapi.tiangolo.com/
Tutorial - User Guide：https://fastapi.tiangolo.com/tutorial/
Path / Query / Body：Path Params、Query Params、Body
Dependencies：https://fastapi.tiangolo.com/tutorial/dependencies/
Security / JWT：https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/
Bigger Applications (Routers)：https://fastapi.tiangolo.com/tutorial/bigger-applications/
CORS：https://fastapi.tiangolo.com/tutorial/cors/
Testing：https://fastapi.tiangolo.com/tutorial/testing/
Deployment：https://fastapi.tiangolo.com/deployment/
Pydantic v2：https://docs.pydantic.dev/
SQLAlchemy 2.0：https://docs.sqlalchemy.org/
Uvicorn：https://www.uvicorn.org/
Starlette：https://www.starlette.io/

把 Router、Pydantic、依赖注入、笔记 CRUD 与 JWT 登录串起来，就是一个可继续长成「个人笔记 API」的骨架。下一步通常是：Alembic 迁移、刷新令牌与权限角色、PostgreSQL、后台任务（BackgroundTasks / Celery / ARQ），以及按官方 Deployment换成真正的进程与反向代理模型。