☁️☁️从零到一的 FastAPI 入门教程

Application
约 52 分钟阅读

阅读量:69次
FastAPI
实现一个简单但工程化到位的「待办事项（Todo）API」。涵盖路由、参数验证、依赖注入、数据库集成（SQLModel +SQLite/aiosqlite）以及全链路异步。

项目简介

  • 路由与参数验证：Path、Query、Body 的校验与约束
  • 依赖注入（DI）：数据库会话、分页入参、对象获取等可复用依赖
  • 数据库集成（SQLModel）：建模、异步会话（AsyncSession）、查询与事务
  • 异步编程：全链路 async/await、启动生命周期（lifespan）异步建表
  • 工程化结构：分包、路由分层、响应模型、错误处理、自动文档

API 一览

  • POST /api/v1/todos：创建待办
  • GET /api/v1/todos：列表（分页、按完成状态与关键词过滤）
  • GET /api/v1/todos/{todo_id}：获取详情
  • PATCH /api/v1/todos/{todo_id}：部分更新（标题/描述/完成状态）
  • DELETE /api/v1/todos/{todo_id}：删除
  • 自带：OpenAPI 文档 /docs、/redoc

1. 项目结构（建议）

fastapi-todo/
├─ app/
│  ├─ __init__.py
│  ├─ main.py            # 入口，挂载路由、lifespan 启动/关闭
│  ├─ db.py              # 数据库引擎、会话依赖、建表函数
│  ├─ models.py          # SQLModel 表 & 字段定义
│  ├─ schemas.py         # 输入/输出(pydantic)模型
│  ├─ deps.py            # 通用依赖（分页参数、对象加载等）
│  └─ routers/
│     ├─ __init__.py
│     └─ todos.py        # Todo 路由（CRUD）
├─ requirements.txt
└─ README.md（可选）

2. 环境准备

python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -U pip

# 安装依赖（FastAPI + Uvicorn + SQLModel + aiosqlite）
pip install "fastapi" "uvicorn[standard]" "sqlmodel>=0.0.14" "aiosqlite"

requirements.txt（如果更爱固定清单）：

fastapi
uvicorn[standard]
sqlmodel>=0.0.14
aiosqlite

3. 代码实现

3.1 app/models.py —— 定义数据表

# app/models.py
from __future__ import annotations
from datetime import datetime
from typing import Optional
from sqlmodel import SQLModel, Field

class Todo(SQLModel, table=True):
    """
    SQLModel 既是 ORM 模型，也是 Pydantic 模型。
    """
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(min_length=1, max_length=100, description="标题")
    description: Optional[str] = Field(default=None, max_length=1000, description="描述")
    completed: bool = Field(default=False, description="是否完成")
    created_at: datetime = Field(default_factory=datetime.utcnow, nullable=False)
    updated_at: Optional[datetime] = Field(default=None)

说明

  • Field(min_length=..., max_length=...) 直接约束输入长度（Pydantic 校验）。
  • created_at 用 default_factory 自动赋值；updated_at 在更新接口里维护。

3.2 app/schemas.py —— 定义输入/输出模型

# app/schemas.py
from __future__ import annotations
from typing import Optional
from pydantic import BaseModel, Field

class TodoCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = Field(default=None, max_length=1000)

class TodoUpdate(BaseModel):
    title: Optional[str] = Field(default=None, min_length=1, max_length=100)
    description: Optional[str] = Field(default=None, max_length=1000)
    completed: Optional[bool] = None

class TodoRead(BaseModel):
    id: int
    title: str
    description: Optional[str]
    completed: bool
    # 用字符串而非 datetime，避免前端解析负担
    created_at: str
    updated_at: Optional[str]

说明

  • 将“写入模型”（Create/Update）与“读出模型”（Read）分离是实战最佳实践。
  • 输出中把 datetime 转为 str 便于前端直接展示（也可保留为 datetime，由客户端解析）。

3.3 app/db.py —— 异步引擎与会话依赖

# app/db.py
from __future__ import annotations
from typing import AsyncGenerator
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./todo.db"

engine: AsyncEngine = create_async_engine(
    DATABASE_URL,
    echo=False,         # 调试时可 True 看 SQL
    future=True
)

# AsyncSession 工厂
AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# 依赖：按请求作用域提供会话
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session

# 启动时建表（异步）
async def init_db() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

# 关闭时释放
async def close_db() -> None:
    await engine.dispose()

3.4 app/deps.py —— 通用依赖（分页、对象加载）

# app/deps.py
from __future__ import annotations
from typing import Optional
from fastapi import Depends, HTTPException, Path, Query, status
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from app.db import get_session
from app.models import Todo

# 分页与过滤参数
def pagination_params(
    page: int = Query(1, ge=1, description="页码，从 1 开始"),
    size: int = Query(20, ge=1, le=100, description="每页数量 1~100"),
    completed: Optional[bool] = Query(default=None, description="按完成状态过滤"),
    q: Optional[str] = Query(default=None, min_length=1, max_length=100, description="标题关键词")
):
    return {"page": page, "size": size, "completed": completed, "q": q}

# 加载 Todo 对象的依赖（避免每个路由重复写）
async def get_todo_or_404(
    todo_id: int = Path(..., ge=1),
    session: AsyncSession = Depends(get_session)
) -> Todo:
    todo = await session.get(Todo, todo_id)
    if not todo:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found")
    return todo

3.5 app/routers/todos.py —— 业务路由（CRUD）

# app/routers/todos.py
from __future__ import annotations
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import select, col
from sqlmodel.ext.asyncio.session import AsyncSession

from app.db import get_session
from app.models import Todo
from app.schemas import TodoCreate, TodoRead, TodoUpdate
from app.deps import pagination_params, get_todo_or_404

router = APIRouter(prefix="/api/v1/todos", tags=["todos"])

def to_read_model(todo: Todo) -> TodoRead:
    return TodoRead(
        id=todo.id,
        title=todo.title,
        description=todo.description,
        completed=todo.completed,
        created_at=todo.created_at.isoformat(),
        updated_at=todo.updated_at.isoformat() if todo.updated_at else None,
    )

@router.post("", response_model=TodoRead, status_code=status.HTTP_201_CREATED)
async def create_todo(
    payload: TodoCreate,
    session: AsyncSession = Depends(get_session)
) -> TodoRead:
    todo = Todo(title=payload.title, description=payload.description)
    session.add(todo)
    await session.commit()
    await session.refresh(todo)
    return to_read_model(todo)

@router.get("", response_model=List[TodoRead])
async def list_todos(
    params: Dict[str, Any] = Depends(pagination_params),
    session: AsyncSession = Depends(get_session)
) -> List[TodoRead]:
    page = params["page"]
    size = params["size"]
    completed: Optional[bool] = params["completed"]
    q: Optional[str] = params["q"]

    stmt = select(Todo)
    if completed is not None:
        stmt = stmt.where(Todo.completed == completed)
    if q:
        # SQLite 大小写不敏感 LIKE，可换成 ilike（在 PG 上更好）
        stmt = stmt.where(col(Todo.title).like(f"%{q}%"))

    stmt = stmt.order_by(Todo.created_at.desc()).offset((page - 1) * size).limit(size)
    result = await session.exec(stmt)
    todos = result.all()
    return [to_read_model(t) for t in todos]

@router.get("/{todo_id}", response_model=TodoRead)
async def get_todo(todo: Todo = Depends(get_todo_or_404)) -> TodoRead:
    return to_read_model(todo)

@router.patch("/{todo_id}", response_model=TodoRead)
async def update_todo(
    updates: TodoUpdate,
    session: AsyncSession = Depends(get_session),
    todo: Todo = Depends(get_todo_or_404)
) -> TodoRead:
    # 仅更新有传值的字段
    dirty = False
    if updates.title is not None:
        todo.title = updates.title
        dirty = True
    if updates.description is not None:
        todo.description = updates.description
        dirty = True
    if updates.completed is not None:
        todo.completed = updates.completed
        dirty = True

    if not dirty:
        # 没有任何字段被更新
        return to_read_model(todo)

    todo.updated_at = datetime.utcnow()
    session.add(todo)
    await session.commit()
    await session.refresh(todo)
    return to_read_model(todo)

@router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(
    session: AsyncSession = Depends(get_session),
    todo: Todo = Depends(get_todo_or_404)
) -> None:
    await session.delete(todo)
    await session.commit()
    return

亮点

  • 列表接口支持分页 + 条件过滤 + 关键词搜索。
  • get_todo_or_404 把「按 ID 获取并 404」抽成依赖，复用在 GET/PATCH/DELETE。
  • PATCH 采用部分更新语义，避免客户端必须传全量字段。
  • 全部端点是 async def，数据库操作通过 await 调用，真正异步链路。

3.6 app/main.py —— 应用入口与 lifespan

# app/main.py
from __future__ import annotations
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.db import init_db, close_db
from app.routers.todos import router as todos_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 应用启动：建表
    await init_db()
    yield
    # 应用关闭：释放连接
    await close_db()

app = FastAPI(
    title="Todo API (FastAPI + SQLModel)",
    version="1.0.0",
    lifespan=lifespan
)

# CORS（视需要允许前端域名）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],   # 生产环境请改成具体域
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 路由挂载
app.include_router(todos_router)

4. 运行与验证

启动服务

uvicorn app.main:app --reload

打开浏览器：

  • 交互式文档（Swagger UI）：http://127.0.0.1:8000/docs
  • Redoc 文档：http://127.0.0.1:8000/redoc

cURL 快速验收

# 创建
curl -X POST http://127.0.0.1:8000/api/v1/todos \
  -H "Content-Type: application/json" \
  -d '{"title":"写 FastAPI 教程","description":"含 SQLModel + 异步"}'

# 列表（分页 + 过滤）
curl "http://127.0.0.1:8000/api/v1/todos?page=1&size=10&completed=false&q=FastAPI"

# 获取详情
curl http://127.0.0.1:8000/api/v1/todos/1

# 部分更新
curl -X PATCH http://127.0.0.1:8000/api/v1/todos/1 \
  -H "Content-Type: application/json" \
  -d '{"completed":true,"title":"写完教程"}'

# 删除
curl -X DELETE http://127.0.0.1:8000/api/v1/todos/1

5. 核心知识点拆解

5.1 路由与参数验证

  • Path 参数：/{todo_id} 用 Path(ge=1) 限制为正整数。
  • Query 参数：分页 page/size，限制范围避免大页导致压力。
  • Body 参数：TodoCreate/TodoUpdate 校验长度、可空性。
  • 响应模型（response_model）：保证输出结构稳定、可被前端/文档依赖。

5.2 依赖注入（DI）

  • Depends(get_session)：按请求生成独立的 AsyncSession，自动回收。
  • get_todo_or_404：将「对象加载 + 404」封装成依赖，路由逻辑更简洁。
  • pagination_params：把常用 Query 参数组合为一个依赖对象。

5.3 SQLModel + 异步

  • create_async_engine("sqlite+aiosqlite:///...") 让 SQLite 在事件循环中工作。
  • AsyncSession + await session.exec(select(...))：纯异步数据库访问。
  • lifespan 异步启动：await init_db() 保证首次运行时自动建表。

5.4 错误处理与返回码

  • 找不到资源 → 404；创建成功 → 201；删除成功 → 204。
  • 统一通过 HTTPException(status_code, detail="...") 抛出业务错误。

6. 进一步扩展（建议顺序）

  1. 总数统计与分页元信息：GET /todos/meta 返回 total/count_completed。
  2. 排序与多字段过滤：支持按 created_at/updated_at 排序。
  3. 鉴权：加一个 get_current_user 依赖（JWT 或简易 API-Key），在表里新增 owner_id 实现用户隔离。
  4. 迁移管理：接入 Alembic（SQLAlchemy 2.0 兼容）做 schema 进化。
  5. PostgreSQL：把连接串改为 postgresql+asyncpg://... 即可切到生产级数据库。
  6. 测试：用 httpx.AsyncClient + pytest 编写异步集成测试，搭配 pytest-asyncio。

7. 常见坑位与排查

  • 同步/异步混用：确保所有路由与 DB 调用都是 async/await，不要把 create_engine（同步）误用在异步项目里。
  • SQLite 并发：开发足够，生产建议用 PostgreSQL；并发写多时更稳。
  • 会话过期：expire_on_commit=False 能避免 commit 后属性失效需要再次访问数据库。
  • LIKE/ILIKE 差异：示例里用 like；如果换 PG，改 ilike 以支持大小写不敏感。

8. 你可以直接复制的最小可运行片段（单文件版，可快速试跑）

如果你想先“跑起来再拆包”，可以用下面的单文件示例验证后再回到分层版本。
# run.py
from __future__ import annotations
from contextlib import asynccontextmanager
from datetime import datetime
from typing import Optional, List

from fastapi import FastAPI, Depends, HTTPException, Path, Query, status
from fastapi.middleware.cors import CORSMiddleware
from sqlmodel import SQLModel, Field, select, col
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker

class Todo(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(min_length=1, max_length=100)
    description: Optional[str] = Field(default=None, max_length=1000)
    completed: bool = Field(default=False)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: Optional[datetime] = None

DATABASE_URL = "sqlite+aiosqlite:///./todo.db"
engine = create_async_engine(DATABASE_URL, echo=False, future=True)
AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)

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

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    yield
    await engine.dispose()

app = FastAPI(title="Todo API", version="1.0.0", lifespan=lifespan)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"]
)

@app.post("/api/v1/todos", status_code=201)
async def create_todo(
    title: str,
    description: Optional[str] = None,
    session: AsyncSession = Depends(get_session)
):
    todo = Todo(title=title, description=description)
    session.add(todo)
    await session.commit()
    await session.refresh(todo)
    return todo

@app.get("/api/v1/todos")
async def list_todos(
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
    completed: Optional[bool] = None,
    q: Optional[str] = Query(default=None, min_length=1, max_length=100),
    session: AsyncSession = Depends(get_session)
) -> List[Todo]:
    stmt = select(Todo)
    if completed is not None:
        stmt = stmt.where(Todo.completed == completed)
    if q:
        stmt = stmt.where(col(Todo.title).like(f"%{q}%"))
    stmt = stmt.order_by(Todo.created_at.desc()).offset((page - 1) * size).limit(size)
    result = await session.exec(stmt)
    return result.all()

@app.get("/api/v1/todos/{todo_id}")
async def get_todo(todo_id: int = Path(..., ge=1), session: AsyncSession = Depends(get_session)):
    todo = await session.get(Todo, todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo

@app.patch("/api/v1/todos/{todo_id}")
async def update_todo(
    todo_id: int = Path(..., ge=1),
    title: Optional[str] = Query(default=None, min_length=1, max_length=100),
    description: Optional[str] = Query(default=None, max_length=1000),
    completed: Optional[bool] = None,
    session: AsyncSession = Depends(get_session)
):
    todo = await session.get(Todo, todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    if title is not None:
        todo.title = title
    if description is not None:
        todo.description = description
    if completed is not None:
        todo.completed = completed
    todo.updated_at = datetime.utcnow()
    session.add(todo)
    await session.commit()
    await session.refresh(todo)
    return todo

@app.delete("/api/v1/todos/{todo_id}", status_code=204)
async def delete_todo(todo_id: int = Path(..., ge=1), session: AsyncSession = Depends(get_session)):
    todo = await session.get(Todo, todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    await session.delete(todo)
    await session.commit()
    return

运行：

uvicorn run:app --reload

添加新评论