FastAPI 构建 Web API 入门指南
FastAPI 入门实战:用 SQLModel 连数据库、Pydantic 做校验,实现任务管理 API 的完整 CRUD(POST/GET/PUT/DELETE),自动获得交互式文档。
直接回答:FastAPI 用类型注解驱动一切——请求校验、响应序列化、自动文档一次搞定。配合 SQLModel 操作数据库,不到百行代码就能交付一个带交互式文档的完整 CRUD API。
为什么是 FastAPI
- 类型即契约:参数校验、文档生成全来自类型注解;
- 异步原生:高并发 IO 场景吞吐出色;
- 自动文档:Swagger UI 开箱即用,便于查看与试用接口。
搭建项目
pip install "fastapi[standard]" sqlmodel
配置数据库(SQLModel)
SQLModel 是 FastAPI 作者的作品:一个类同时是 Pydantic 模型与 SQLAlchemy 表。
# database.py
from sqlmodel import SQLModel, Field, create_engine, Session
class Task(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
title: str
done: bool = False
class TaskInput(SQLModel):
title: str = Field(min_length=1, max_length=200)
done: bool = False
engine = create_engine("sqlite:///tasks.db", connect_args={"check_same_thread": False})
SQLModel.metadata.create_all(engine)
POST:创建任务
# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select
from database import engine, Task, TaskInput
app = FastAPI()
def get_session():
with Session(engine) as s:
yield s
@app.post("/tasks", response_model=Task, status_code=201)
def create_task(data: TaskInput, session: Session = Depends(get_session)):
task = Task.model_validate(data)
session.add(task)
session.commit()
session.refresh(task)
return task
请求体使用独立非表模型 TaskInput 校验,避免接受客户端主键;上线前需添加认证及对象级权限——缺字段、类型错,422 响应自动返回,零手写校验代码。
GET:查询任务
@app.get("/tasks", response_model=list[Task])
def list_tasks(done: bool | None = None, session: Session = Depends(get_session)):
stmt = select(Task)
if done is not None:
stmt = stmt.where(Task.done == done)
return session.exec(stmt.limit(100)).all()
@app.get("/tasks/{task_id}", response_model=Task)
def get_task(task_id: int, session: Session = Depends(get_session)):
task = session.get(Task, task_id)
if not task:
raise HTTPException(404, "任务不存在")
return task
查询参数 done 声明为可选布尔,FastAPI 自动从 querystring 解析转换。
PUT:更新任务
@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, data: TaskInput, session: Session = Depends(get_session)):
task = session.get(Task, task_id)
if not task:
raise HTTPException(404, "任务不存在")
task.title = data.title
task.done = data.done
session.commit()
return task
DELETE:删除任务
@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int, session: Session = Depends(get_session)):
task = session.get(Task, task_id)
if not task:
raise HTTPException(404, "任务不存在")
session.delete(task)
session.commit()
跑起来看文档
fastapi dev main.py
打开 http://localhost:8000/docs——Swagger UI 上直接点试每个端点,请求响应模型一目了然。文档由声明生成;仍需验证业务规则。create_all 仅用于本地起步,生产结构变更使用迁移;列表的 100 条上限需进一步设计分页。
常见问题(FAQ)
Q:SQLModel 和原生 SQLAlchemy 怎么选?
A:中小项目 SQLModel 最顺手(模型即表即 Schema);复杂查询、存量 SQLAlchemy 生态(Alembic 迁移等)项目直接用 SQLAlchemy,Pydantic 只做接口层。
Q:生产环境怎么跑?
A:uvicorn main:app --host 0.0.0.0 --workers 4 或容器化部署;记得关掉文档或加保护(/docs 暴露接口结构)。
Q:怎么做接口版本管理?
A:按路径前缀拆分路由器:app.include_router(v1, prefix="/api/v1")。破坏性变更开 v2 并行运行,客户端渐进迁移。
官方参考
本文基于官方文档整理,未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全;局部片段不等同于完整生产应用。