1. 什么是 FastAPI? #
FastAPI 是一个用于构建 Web API 的 Python 框架,特点是快、简单、自动生成文档。
你写几行 Python 代码,就能得到一个可以通过浏览器或程序访问的接口。比如你写一个「获取用户信息」的函数,FastAPI 会把它变成 GET /users/1 这样的网址,别人访问就能拿到数据。
打个比方:就像开一家「自助取餐窗口」——你定义好菜单(接口),顾客按编号点餐(发请求),窗口自动把菜端出来(返回数据)。FastAPI 帮你把「菜单」和「取餐流程」都标准化好了。
2. 前置知识:Web API 与 HTTP 方法 #
在学 FastAPI 之前,需要了解两个基础概念。
2.1 什么是 Web API? #
API(Application Programming Interface)是程序之间通信的接口。Web API 就是通过网址(URL)来调用的接口。
例如:访问 https://api.example.com/users/1,服务器返回用户 1 的信息(通常是 JSON 格式)。你的程序发请求,服务器返回数据,这就是 Web API 的典型用法。
2.2 常用 HTTP 方法 #
| 方法 | 含义 | 常见用途 |
|---|---|---|
| GET | 获取 | 查询数据,如获取用户列表 |
| POST | 提交 | 创建数据,如新建用户 |
| PUT | 更新 | 整体替换,如修改用户信息 |
| DELETE | 删除 | 删除数据 |
FastAPI 用装饰器区分这些方法:@app.get()和@app.post()等。
3. 安装 #
FastAPI 需要配合 ASGI 服务器 才能运行,推荐使用 Uvicorn。
pip install fastapi uvicorn[standard]若使用 uv 管理依赖:uv add fastapi "uvicorn[standard]"
4. 第一个 FastAPI 应用 #
# 导入 FastAPI 类
from fastapi import FastAPI
# 导入 uvicorn 用于启动服务器
import uvicorn
# 创建 FastAPI 应用实例,title 会显示在自动文档中
app = FastAPI(title="我的第一个 API")
# 使用 @app.get 定义 GET 请求的路由,"/" 表示根路径
@app.get("/")
def read_root():
# 返回一个字典,FastAPI 会自动转为 JSON
return {"message": "Hello, FastAPI!"}
# 定义 GET /items/{item_id},item_id 是路径参数
@app.get("/items/{item_id}")
def read_item(item_id: int):
# 路径参数会自动解析并做类型校验
return {"item_id": item_id, "name": f"商品{item_id}"}
# 程序入口:直接运行此文件时执行
if __name__ == "__main__":
# 启动服务器,host 和 port 为监听地址和端口
uvicorn.run(app, host="127.0.0.1", port=8000)运行方式:保存为 main.py,执行 python main.py,然后访问:
http://127.0.0.1:8000→ 返回{"message": "Hello, FastAPI!"}http://127.0.0.1:8000/items/1→ 返回{"item_id": 1, "name": "商品1"}http://127.0.0.1:8000/docs→ 自动生成的交互式文档
5. 路径参数与查询参数 #
5.1 路径参数 #
路径参数写在 URL 路径中,如 /items/123 里的 123。
# 路径参数:写在花括号 {param} 中,与函数参数同名
@app.get("/users/{user_id}")
def get_user(user_id: int):
return {"user_id": user_id}5.2 查询参数 #
查询参数写在 URL 的 ? 后面,如 /items?skip=0&limit=10。
# 查询参数:有默认值的函数参数,未在路径中出现的即为查询参数
@app.get("/items")
def list_items(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}访问 http://127.0.0.1:8000/items?skip=5&limit=20 会得到 {"skip": 5, "limit": 20}。
6. 请求体与 Pydantic 模型 #
6.1 Pydantic 是什么? #
Pydantic 是 FastAPI 用来做数据校验和序列化的库。你定义一个「模型」(类),指定每个字段的类型,Pydantic 会自动:
- 校验传入数据是否符合类型
- 把 Python 对象转成 JSON
- 生成 API 文档中的参数说明
6.2 定义模型并接收请求体 #
当客户端用 POST 发送 JSON 数据时,可以用 Pydantic 模型来接收和校验。
# 导入 FastAPI 和 Pydantic 的 BaseModel
from fastapi import FastAPI
from pydantic import BaseModel
# 创建应用
app = FastAPI()
# 定义请求体模型:继承 BaseModel,声明字段及类型
class Item(BaseModel):
name: str
price: float
is_offer: bool = False
# POST 请求:item 会自动从请求体中解析,并做类型校验
@app.post("/items/")
def create_item(item: Item):
return item
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)curl -i -X POST "http://127.0.0.1:8000/items/" -H "Content-Type: application/json" -d '{"name":"apple","price":12.5}'7. 增删改查(CRUD) #
# 导入 FastAPI 和 Pydantic
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
# 创建应用
app = FastAPI(title="商品 API")
# 内存中模拟数据库(实际项目会用真实数据库)
items_db = {}
next_id = 1
# 定义商品模型
class Item(BaseModel):
name: str
price: float
description: str = ""
# 获取所有商品
@app.get("/items")
def list_items():
return {"items": [{"id": k, **v.model_dump()} for k, v in items_db.items()]}
# 根据 id 获取单个商品
@app.get("/items/{item_id}")
def get_item(item_id: int):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="商品不存在")
return {"id": item_id, **items_db[item_id].model_dump()}
# 创建商品(id 自动递增)
@app.post("/items")
def create_item(item: Item):
global next_id
items_db[next_id] = item
result_id = next_id
next_id += 1
return {"message": "创建成功", "id": result_id, "item": item}
# 更新商品
@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="商品不存在")
items_db[item_id] = item
return {"message": "更新成功", "id": item_id, "item": item}
# 删除商品
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="商品不存在")
del items_db[item_id]
return {"message": "删除成功"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)运行方式:保存为 main.py,执行 python main.py,通过 /docs 或 curl 测试各接口。
# 1) GET /items 获取所有商品
curl -i "http://127.0.0.1:8000/items"# 2) GET /items/{item_id} 获取单个商品(示例 id=1)
curl -i "http://127.0.0.1:8000/items/1"# 3) POST /items 创建商品
curl -i -X POST "http://127.0.0.1:8000/items" \
-H "Content-Type: application/json" \
-d '{"name":"apple","price":12.5,"description":"fresh"}'# 4) PUT /items/{item_id} 更新商品(示例 id=1)
curl -i -X PUT "http://127.0.0.1:8000/items/1" \
-H "Content-Type: application/json" \
-d '{"name":"apple-pro","price":19.9,"description":"updated"}'# 5) DELETE /items/{item_id} 删除商品(示例 id=1)
curl -i -X DELETE "http://127.0.0.1:8000/items/1"补充:这组命令里 GET/PUT/DELETE 用到 id=1,所以建议先执行一次 POST 创建数据,再测其余 3 条。
8. 依赖注入(Depends) #
8.1 什么是依赖注入? #
依赖注入 是把「获取某样东西」的逻辑单独写成函数,由 FastAPI 在每次请求时自动调用并传入路由。这样路由函数只关心业务逻辑,不必重复写「取参数、校验、获取数据库连接」等代码。
通俗理解:就像点外卖——你只说「我要一份盖浇饭」,平台自动帮你联系餐厅、安排骑手。你不需要自己打电话、自己取餐。Depends 就是 FastAPI 里的「自动安排」机制。
8.2 基础用法 #
用 Depends(函数名) 声明依赖,FastAPI 会在处理请求前先调用该函数,把返回值传给路由参数。
# 导入 FastAPI 和 Depends
from fastapi import FastAPI, Depends
# 创建应用
app = FastAPI()
# 定义依赖函数:返回公共的查询参数
def common_params(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
# 使用 Depends 注入:commons 会自动调用 common_params 并传入结果
@app.get("/items")
def read_items(commons: dict = Depends(common_params)):
return commons
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)访问 http://127.0.0.1:8000/items?skip=5&limit=20 会得到 {"skip": 5, "limit": 20}。
8.3 依赖链:依赖可以依赖其他依赖 #
依赖函数本身也可以依赖其他依赖,形成依赖链。FastAPI 会按依赖关系自动解析并调用。
# 导入 FastAPI 和 Depends
from fastapi import FastAPI, Depends
# 创建应用
app = FastAPI()
# 第一层依赖:分页参数
def pagination(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
# 第二层依赖:依赖 pagination,在其基础上增加 keyword
def search_params(pg: dict = Depends(pagination), keyword: str = None):
return {**pg, "keyword": keyword or ""}
# 路由依赖 search_params(内部会先调用 pagination)
@app.get("/items")
def list_items(params: dict = Depends(search_params)):
return params
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)访问 http://127.0.0.1:8000/items?skip=2&limit=5&keyword=手机 会得到 {"skip": 2, "limit": 5, "keyword": "手机"}。
8.4 常见场景:模拟「当前用户校验」 #
依赖注入常用于校验请求头、Token 等,通过后再把用户信息传给路由。从请求头读取参数需使用 Header。
# 导入 FastAPI、Depends、Header 和 HTTPException
from fastapi import FastAPI, Depends, Header, HTTPException
# 创建应用
app = FastAPI()
# 从请求头 X-User-Id 读取用户 id,Header(None) 表示可选,缺省为 None
def get_current_user_id(x_user_id: str = Header(None)):
if not x_user_id:
raise HTTPException(status_code=401, detail="请提供 X-User-Id 请求头")
return {"user_id": x_user_id}
# 需要「登录」的接口:依赖 get_current_user_id
@app.get("/me")
def read_me(user: dict = Depends(get_current_user_id)):
return {"message": f"当前用户: {user['user_id']}"}
# 公开接口:不声明依赖
@app.get("/public")
def read_public():
return {"message": "公开接口"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)测试方式:
# 不带请求头,返回 401
curl -i "http://127.0.0.1:8000/me"
# 带 X-User-Id,返回 200
curl -i -H "X-User-Id: alice" "http://127.0.0.1:8000/me"8.6 常见场景:数据库连接 #
通过依赖注入管理数据库连接,每次请求获取连接、用完后自动关闭,避免连接泄漏。使用 yield 时,yield 之前的代码在请求开始时执行,之后的代码在请求结束后执行(类似 finally)。
# 导入 asynccontextmanager 上下文管理器用于 lifespan 生命周期管理
from contextlib import asynccontextmanager
# 从 FastAPI 导入 FastAPI 应用类和 Depends 依赖注入工具
from fastapi import FastAPI, Depends
# 导入 sqlite3 模块,用于操作 SQLite 数据库
import sqlite3
# 定义数据库文件路径
DB_PATH = "demo.db"
# 依赖函数:获取数据库连接,使用 yield 实现请求结束后自动关闭连接
def get_db():
# 连接到 SQLite 数据库
conn = sqlite3.connect(DB_PATH)
# 设置查询返回的每一行以 dict 形式表现,支持通过列名获取数据
conn.row_factory = sqlite3.Row
try:
# 通过 yield 暴露连接对象
yield conn
finally:
# 请求结束后关闭连接,防止资源泄漏
conn.close()
# 初始化数据库,创建 users 表并插入初始用户数据
def init_db():
# 建立数据库连接
conn = sqlite3.connect(DB_PATH)
# 创建表(如果不存在),包含 id 和 name 字段
conn.execute("CREATE TABLE IF NOT EXISTS users (id INT, name TEXT)")
# 插入或更新样例数据:id=1 的张三、id=2 的李四
conn.execute("INSERT OR REPLACE INTO users VALUES (1, '张三'), (2, '李四')")
# 提交事务
conn.commit()
# 关闭连接
conn.close()
# 声明异步 lifespan 生命周期上下文,在应用启动时初始化数据库
@asynccontextmanager
async def lifespan(app: FastAPI):
# 在应用启动阶段调用数据库初始化函数
init_db()
# 进入主生命周期,后续事件继续执行
yield
# 创建 FastAPI 应用实例,并指定 lifespan 生命周期管理
app = FastAPI(lifespan=lifespan)
# 定义获取单个用户信息的 GET 路由接口,依赖数据库连接
@app.get("/users/{user_id}")
def get_user(user_id: int, conn=Depends(get_db)):
# 执行 SQL 查询,根据用户 id 查找用户
cur = conn.execute("SELECT id, name FROM users WHERE id = ?", (user_id,))
# 获取一条查询结果
row = cur.fetchone()
# 如果没有找到对应用户,返回错误信息
if not row:
return {"error": "用户不存在"}
# 找到对应用户,返回 id 和 name 信息
return {"id": row["id"], "name": row["name"]}
# 判断当前脚本是否为主程序入口
if __name__ == "__main__":
# 导入 uvicorn 用于启动 FastAPI 应用
import uvicorn
# 启动 uvicorn 服务,监听本地 127.0.0.1:8000 端口
uvicorn.run(app, host="127.0.0.1", port=8000)运行方式:保存为 main.py,执行 python main.py,访问 http://127.0.0.1:8000/users/1 可得到 {"id": 1, "name": "张三"}。
8.7 小结 #
| 要点 | 说明 |
|---|---|
| 声明方式 | 参数 = Depends(依赖函数) |
| 执行时机 | 每次请求前自动调用依赖函数 |
| 依赖链 | 依赖函数可以再依赖其他依赖 |
| 常见用途 | 分页参数、用户校验、数据库连接等 |
9. CORS 跨域配置 #
当前端(如网页)从不同域名访问你的 API 时,浏览器会做跨域检查。若未配置 CORS,请求可能被拦截。
# 导入 FastAPI 类
from fastapi import FastAPI
# 导入用于支持跨域请求的 CORS 中间件
from fastapi.middleware.cors import CORSMiddleware
# 创建 FastAPI 应用实例
app = FastAPI()
# 配置并添加 CORS 中间件,允许所有来源的跨域请求
app.add_middleware(
CORSMiddleware,
# 允许所有来源访问,生产环境推荐填写具体域名
allow_origins=["*"],
# 允许客户端发送 cookies 等凭证信息
allow_credentials=True,
# 允许所有 HTTP 方法
allow_methods=["*"],
# 允许所有请求头
allow_headers=["*"],
)
# 定义根路径的 GET 请求接口
@app.get("/")
def read_root():
# 返回一条消息
return {"message": "Hello"}
# 判断是否以脚本方式运行
if __name__ == "__main__":
# 导入 uvicorn,用于运行 ASGI 服务器
import uvicorn
# 启动 FastAPI 应用,监听本地 127.0.0.1:8000
uvicorn.run(app, host="127.0.0.1", port=8000)10. 自动文档 #
FastAPI 会自动根据你的路由和模型生成文档,无需额外配置。
| 地址 | 说明 |
|---|---|
http://127.0.0.1:8000/docs |
Swagger UI,可在线调试 |
http://127.0.0.1:8000/redoc |
ReDoc,阅读式文档 |
11. 运行方式 #
方式一:在代码中调用 uvicorn(如前文示例)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)方式二:命令行启动(推荐开发时使用 --reload 热重载)
uvicorn main:app --reload --host 127.0.0.1 --port 8000其中 main 是文件名,app 是 FastAPI 实例的变量名。
12. 总结 #
- FastAPI 用装饰器定义路由,用 Pydantic 做请求体验证,用
Depends做依赖注入。 - 路径参数 写在
{param}中,查询参数 用带默认值的函数参数。 - 请求体 用 Pydantic 模型接收,自动校验并生成文档。
- 启动后访问
/docs可查看和调试所有接口。