当前位置：首页 > news >正文

FastAPI—学习1

news 2025/7/1 8:37:58

1. 基础代码

from fastapi import FastAPI  # 导入 FastAPI 类# 创建 FastAPI 应用实例
app = FastAPI()# 定义一个 GET 请求的接口，路径为 /items/{item_id}
# {item_id} 是路径参数，用户可以在 URL 中传递不同的 item_id
@app.get("/items/{item_id}") 
def read_item(item_id):"""读取物品信息的接口参数:item_id: 路径参数，表示物品的ID返回:一个包含 item_id 的字典"""return {"item_id": item_id}  # 返回一个字典，键为 item_id，值为传入的参数

代码讲解

导入 FastAPI：from fastapi import FastAPI
创建应用实例：app = FastAPI()，这是所有 FastAPI 项目的基础。
定义路由和接口：@app.get("/items/{item_id}") 表示当用户访问 /items/123 这样的路径时，会调用下面的 read_item 函数。
路径参数：item_id 是从 URL 路径中获取的参数。
返回值：返回一个字典，FastAPI 会自动将其转换为 JSON 格式返回给前端。

uvicorn main:app --reload

在浏览器访问 http://127.0.0.1:8000/items/123，看到返回结果了。

2. 查询参数和 POST 请求

2.1 查询参数（Query Parameters）

除了路径参数，接口还可以接收查询参数。例如：

访问 http://127.0.0.1:8000/items/123?name=apple&price=5，其中 name 和 price 就是查询参数。查询参数是可选的，没传就会是 None

下面是代码示例和注释：

from fastapi import FastAPIapp = FastAPI()# 路径参数 item_id，查询参数 name 和 price
@app.get("/items/{item_id}")
def read_item(item_id: int, name: str = None, price: float = None):"""读取物品信息接口，支持查询参数参数:item_id: 路径参数，物品IDname: 查询参数，物品名称（可选）price: 查询参数，物品价格（可选）返回:包含 item_id、name、price 的字典"""return {"item_id": item_id, "name": name, "price": price}

测试结果

2.2 POST 请求和请求体（Request Body）

有时候我们需要通过 POST 请求提交数据（比如 JSON），这时可以用 Pydantic 模型来定义数据结构。

from fastapi import FastAPI
from pydantic import BaseModel  # 导入 Pydantic 的 BaseModelapp = FastAPI()# 定义请求体的数据结构
class Item(BaseModel):name: strprice: floatdescription: str = None  # 可选字段# 定义 POST 接口，接收 JSON 数据
@app.post("/items/")
def create_item(item: Item):"""创建物品接口参数:item: 请求体，包含 name、price、description返回:返回接收到的 item 数据"""return {"item": item}

用 Postman 或 curl 发送 POST 请求到 http://127.0.0.1:8000/items/，请求体为：

  {"name": "香蕉","price": 2.5,"description": "新鲜的香蕉"}

测试结果

3. 响应模型/应用路由

这次我们来学习两个非常实用且能显著提升代码质量的功能：

响应模型 (Response Model)：精确控制返回给客户端的数据，避免泄露敏感信息。

应用路由 (APIRouter)：将项目拆分成多个模块，使代码结构更清晰、更易于维护。

3.1 响应模型 (Response Model)

假设你的数据库里存了某个物品的 password 字段，但你绝不想在 API 接口中把它返回给用户。这时，response_model 就派上用场了。

我们可以定义一个专门用于响应的模型，只包含希望公开的字段。

示例代码：

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optionalapp = FastAPI()# 假设这是我们内部使用的数据模型，包含敏感信息
class ItemInternal(BaseModel):name: strprice: floatpassword: str  # 敏感数据，不想返回给用户description: Optional[str] = None# 这是我们定义用来响应的模型，只包含公开字段
class ItemPublic(BaseModel):name: strprice: floatdescription: Optional[str] = None# 模拟一个数据库
fake_db = {"item1": {"name": "苹果", "price": 5.0, "password": "my_secret_password"}
}# 在装饰器中指定 response_model
@app.get("/items/{item_id}", response_model=ItemPublic)
def read_item(item_id: str):"""读取物品信息，但通过 response_model 过滤掉 password 字段"""# 在函数内部，我们仍然可以获取到包含 password 的完整数据internal_data = fake_db.get(item_id)# FastAPI 会自动使用 ItemPublic 模型来过滤返回值return internal_data

当你运行并访问 http://127.0.0.1:8000/items/item1 时，你会发现返回的 JSON 中只包含 name 和 price，password 字段被成功过滤掉了。

测试结果

3.2 应用路由 (APIRouter)

当你的项目越来越大，你会有很多关于用户、物品、订单等的接口。把它们都放在一个 main.py 文件里会显得非常臃肿。APIRouter 可以帮助我们把相关的接口组织到不同的文件中。

让我们来创建一个专门处理物品（items）的路由。

项目结构：

.
├── main.py
└── routers/├── __init__.py  (可以为空文件)└── items.py

第1步：创建 routers/items.py 文件

在这个文件里，我们把所有和 "items" 相关的接口都放进去。

# 文件: routers/items.pyfrom fastapi import APIRouter
from pydantic import BaseModel# 1. 创建一个新的路由实例
router = APIRouter(prefix="/items",  # 2. 给这个路由下的所有路径添加前缀 /itemstags=["Items"],   # 3. 在API文档中，把这些接口分到 "Items" 组里
)class Item(BaseModel):name: strprice: float@router.get("/")
def read_items():return [{"name": "苹果", "price": 5.0}, {"name": "香蕉", "price": 2.5}]@router.get("/{item_id}")
def read_item(item_id: int):return {"name": "苹果", "price": 5.0, "item_id": item_id}@router.post("/")
def create_item(item: Item):return {"message": "创建成功", "item": item}

第2步：修改 main.py 来引入路由

现在，主应用文件 main.py 会变得非常简洁。

# 文件: main.pyfrom fastapi import FastAPI
from routers import items  # 1. 从 routers 包导入 items 模块app = FastAPI()# 2. 将 items.router 包含到主应用中
app.include_router(items.router)@app.get("/")
def read_root():return {"message": "欢迎来到我的商店"}

现在启动应用 (uvicorn main:app --reload)，你可以访问：

http://127.0.0.1:8000/ (来自 main.py)
http://127.0.0.1:8000/items/ (来自 routers/items.py)
http://127.0.0.1:8000/items/123 (来自 routers/items.py)

并且在 API 文档 (/docs) 页面，你会看到接口被清晰地分成了 "default" 和 "Items" 两个组。

4.什么是依赖注入？

简单来说，依赖注入是一种让 FastAPI 帮你处理 "准备工作" 的机制。

你的接口函数可能需要：

一个数据库连接会话。

确认当前用户是否已登录。

一些所有接口都通用的查询参数。

你不需要在每个接口函数里重复写这些逻辑。你只需要把这些逻辑写成一个独立的函数（一个 "依赖")，然后在接口函数里声明：“我需要这个！” FastAPI 就会在调用你的接口前，自动运行这个依赖，并将结果 "注入" 到你的接口函数中。

4.1 简单的依赖：共享通用参数

假设你的很多接口都需要支持分页（skip 和 limit）和搜索（q）。我们可以把这些参数提取到一个依赖函数中。

示例代码：

from typing import Optional
from fastapi import FastAPI, Dependsapp = FastAPI()# 1. 这是我们的依赖函数
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):"""一个处理通用查询参数的依赖。它本身不是一个接口，而是被其他接口调用的“帮手”。"""return {"q": q, "skip": skip, "limit": limit}@app.get("/items/")
# 2. 在接口函数的参数中，使用 Depends 来声明依赖
async def read_items(commons: dict = Depends(common_parameters)):"""这个接口依赖 common_parameters。FastAPI会先调用 common_parameters，然后将返回的字典赋值给 commons 参数。"""# 假设这是你从数据库获取的物品all_items = [{"name": "苹果"}, {"name": "香蕉"}, {"name": "橘子"}]# 我们可以直接使用依赖返回的结果response = {"params": commons, "items": all_items[commons["skip"]:commons["skip"] + commons["limit"]]}return response@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):# 这个接口也复用了同样的依赖，无需重复定义 q, skip, limitreturn {"params": commons, "users": [{"username": "Alice"}, {"username": "Bob"}]}

现在，你可以这样访问：

http://127.0.0.1:8000/items/?skip=1&limit=1
http://127.0.0.1:8000/users/?q=Alice

read_items 和 read_users 都不需要关心 q, skip, limit 是如何被获取和处理的，它们只管接收 commons 这个结果并使用它。代码得到了极大的简化和复用。

4.2 更实用的依赖：用户认证

依赖注入最常见的用途之一就是保护接口，确保只有登录用户才能访问。

示例代码：

from typing import Optional
from fastapi import FastAPI, Depends, Header, HTTPExceptionapp = FastAPI()# 这是一个依赖，用于检查请求头中是否有合法的 Token
async def verify_token(x_token: str = Header(...)):"""检查请求头中的 'X-Token'。- Header(...) 表示这个请求头是必需的。- 如果 Token 不正确，就抛出一个 HTTP 异常，请求会立即中断。"""if x_token != "fake-super-secret-token":raise HTTPException(status_code=401, detail="X-Token header invalid")return x_token# 我们可以在一个接口上组合多个依赖
@app.get("/items_secure/", dependencies=[Depends(verify_token)])
async def read_secure_items():"""这个接口受 verify_token 保护。- dependencies=[...] 这种形式的依赖会运行，但其返回值不会被注入到函数参数中。- 它纯粹用于执行检查或记录日志等操作。"""return [{"item": "Secure Item 1"}, {"item": "Secure Item 2"}]

如何测试：

你可以使用 curl 或 Postman 等工具来测试。

失败的请求 (没有 Token 或 Token 错误)：

设置请求
方法: GET
URL: http://127.0.0.1:8000/items_secure/

添加错误的请求头
点击 Headers 标签页。
添加一个新的键值对：
KEY: X-Token
VALUE: this-is-a-wrong-token (一个随便写的错误值)

成功的请求

通过依赖注入，我们轻松地为任何需要保护的接口添加了认证逻辑，而无需在每个接口内部都写一遍 if token ... else ... 的代码。

依赖注入是 FastAPI 的精髓所在，它让你的应用逻辑和基础设施逻辑（如认证、数据库）解耦。

5.中间件、后台任务

继续深入 FastAPI 的高级特性。这次我们来学习两个非常实用的功能：

中间件 (Middleware)：在每个请求被处理之前和每个响应返回之前执行的通用代码。
后台任务 (Background Tasks)：在返回响应后，在后台异步执行长时间运行的操作

5.1 中间件 (Middleware)

想象一下，你希望：

记录每个 API 请求的处理时间
为所有响应添加一个特定的 HTTP 头
在所有请求到达时验证一个通用的 session

这些跨越多个（甚至所有）接口的通用功能，就是中间件的完美应用场景。中间件就像一个 "检查站"，所有进出的 "交通"（请求和响应）都要经过它。

示例：计算请求处理时间的中间件

让我们来创建一个中间件，它会计算服务器处理每个请求花费了多少时间，并把这个时间放在响应的 X-Process-Time 头里。

import time
from fastapi import FastAPI, Requestapp = FastAPI()# 1. 使用 @app.middleware("http") 装饰器来定义一个中间件
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):"""这个中间件会拦截所有 HTTP 请求。参数:request: 代表当前进来的请求。call_next: 一个函数，它会调用后续的处理程序（即你的路径操作函数）。"""start_time = time.time()  # 记录请求开始的时间# 2. 调用 call_next 来处理请求，并等待获取响应response = await call_next(request)process_time = time.time() - start_time  # 计算处理时间# 3. 在响应的 headers 中添加自定义头response.headers["X-Process-Time"] = str(process_time)return response@app.get("/")
async def main():# 模拟一些处理过程time.sleep(0.5)return {"message": "Hello World"}@app.get("/items/")
async def read_items():# 模拟一些处理过程time.sleep(0.2)return [{"name": "Apple"}, {"name": "Banana"}]

如何测试：

运行上面的代码。
用 Postman 或 curl -v http://127.0.0.1:8000/ 来访问你的接口。
在响应的 Headers部分，你会看到一个新的头，例如 X-Process-Time: 0.50123...。无论你访问哪个接口，这个头都会被添加，这就是中间件的强大之处。

5.2 后台任务 (Background Tasks)

有些操作非常耗时，比如发送邮件、生成报告、处理图片等。如果让用户在浏览器前一直等待这些操作完成，体验会非常糟糕。

后台任务允许你的 API 立即返回响应给用户（例如 "我们已经收到您的请求，正在处理中"），然后在后台继续执行耗时的任务。

示例：发送邮件通知（模拟）

假设用户注册后，我们需要给他发一封欢迎邮件。发送邮件可能会有延迟，我们不希望注册接口因此变慢。

from fastapi import FastAPI, BackgroundTasksapp = FastAPI()# 这是一个模拟的发送邮件函数
def write_notification(email: str, message=""):"""模拟一个耗时的操作，比如发送邮件。这里我们只是把它写入一个文件来模拟。"""import timetime.sleep(5)  # 模拟 5 秒的延迟with open("log.txt", mode="a") as email_file:content = f"发送通知给 {email}: {message}\n"email_file.write(content)@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):"""这个接口会立即返回，然后让后台任务去执行真正的“发送邮件”操作。参数:email: 用户的邮箱地址。background_tasks: FastAPI 会自动注入这个对象。"""# 1. 使用 background_tasks.add_task() 来添加一个后台任务# 参数1: 要执行的函数 (write_notification)# 后续参数: 传递给该函数的参数 (email, message)background_tasks.add_task(write_notification, email, message="欢迎您注册！")# 2. 立即返回响应给用户return {"message": "通知将在后台发送"}

如何测试：

运行代码，并用 Postman 发送一个 POST 请求到 http://127.0.0.1:8000/send-notification/test@example.com。
你会发现，你几乎是瞬间就收到了响应 {"message": "通知将在后台发送"}。
等待大约 5 秒钟，你会发现在你的项目目录下多了一个 log.txt 文件，里面写着：发送通知给 test@example.com: 欢迎您注册！。

这证明了我们的接口没有被耗时操作阻塞，成功地将任务移到了后台执行。

6.文件上传，数据库集成

文件上传 (File Uploads)：让用户可以上传图片、文档等文件到你的服务器。

数据库集成 (Database Integration)：将 FastAPI 与数据库连接，实现数据的持久化存储和管理。这部分内容会比较多，但我会分解成清晰的步骤。

6.1 文件上传 (File Uploads)

FastAPI 处理文件上传非常简单。你需要从 fastapi 导入 File 和 UploadFile。UploadFile 是推荐的方式，因为它更高效，特别是处理大文件时，它会将文件流式传输到内存或磁盘，而不是一次性读入所有内容。

示例：上传一张图片并保存

这个例子会创建一个接口，接收用户上传的文件，并将其保存在服务器的 uploads/ 目录下。

准备工作：

需要安装 python-multipart 库：pip install python-multipart

在你的项目根目录下创建一个名为 uploads 的文件夹。

代码：

import shutil
from pathlib import Path
from fastapi import FastAPI, File, UploadFileapp = FastAPI()# 定义上传文件的保存目录
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)  # 确保目录存在@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):"""接收并保存上传的文件。参数:file: UploadFile 对象，FastAPI 会自动处理从表单接收的文件。File(...) 表示这个字段是必需的。"""# 构造保存路径，使用原始文件名destination = UPLOAD_DIR / file.filenametry:# 使用 shutil.copyfileobj 将上传的文件内容写入到目标文件中# 这是一个高效的方式，适合处理大文件with destination.open("wb") as buffer:shutil.copyfileobj(file.file, buffer)finally:# 确保关闭文件，释放资源file.file.close()return {"filename": file.filename,"content_type": file.content_type,"location": str(destination)}

如何用 Postman 测试：

设置请求

方法: POST

URL: http://127.0.0.1:8000/uploadfile/

设置请求体 (Body)

选择 form-data 标签页。
在 KEY 字段中输入 file（必须和接口函数参数名一致）。
将这一行的类型从 Text 改为 File。
在 VALUE 字段，点击 Select Files 按钮，然后从你的电脑中选择一个文件（比如一张图片）。

发送请求

点击 Send。你会收到一个 JSON 响应，告诉你文件名和保存路径。
检查你项目中的 uploads 文件夹，你会发现文件已经被成功保存了

6.2 数据库集成 (SQLAlchemy)

这是最重要的一步。我们将使用 SQLAlchemy，一个强大的 Python ORM (对象关系映射) 工具，来操作数据库。为了简单起见，我们使用 SQLite 数据库，因为它不需要任何额外的安装和服务，就是一个本地文件。

目标：创建一个完整的、分层的应用来创建和读取 "物品 (Items)"。

准备工作：

安装 SQLAlchemy: pip install sqlalchemy

推荐的项目结构：

为了保持代码清晰，我们将代码拆分成几个文件。

.
├── main.py         # 主应用和 API 接口
├── models.py       # 数据库模型 (数据表结构)
├── schemas.py      # Pydantic 模型 (数据校验和响应)
├── crud.py         # 数据库操作函数 (增删改查)
└── database.py     # 数据库连接和会话设置

第 1 步: database.py - 设置数据库连接

# 文件: database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker# 1. 数据库 URL
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"  # 使用 SQLite 数据库# 2. 创建 SQLAlchemy 引擎
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} # check_same_thread 仅用于 SQLite
)# 3. 创建数据库会话类
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 4. 创建一个 Base 类，之后我们的 ORM 模型会继承它
Base = declarative_base()

2 步: models.py - 定义数据库模型

# 文件: models.py
from sqlalchemy import Column, Integer, String
from .database import Base# 定义一个 Item 模型，它会映射到数据库中的 items 表
class Item(Base):__tablename__ = "items"  # 表名id = Column(Integer, primary_key=True, index=True)name = Column(String, index=True)description = Column(String, index=True)

第 3 步: schemas.py - 定义 Pydantic 模型

这用于请求和响应的数据验证，将 API 数据与数据库模型解耦。

# 文件: schemas.py
from pydantic import BaseModel
from typing import Optional# 用于创建物品的基础模型
class ItemBase(BaseModel):name: strdescription: Optional[str] = None# 创建物品时使用的模型 (继承自 Base)
class ItemCreate(ItemBase):pass# 从数据库读取物品时使用的模型 (包含 id)
class Item(ItemBase):id: intclass Config:from_attributes = True  # Pydantic V2 使用 from_attributes 替代 orm_mode

第 4 步: crud.py - 编写数据库操作函数

CRUD = Create, Read, Update, Delete。

# 文件: crud.py
from sqlalchemy.orm import Session
from . import models, schemas# 读取单个物品
def get_item(db: Session, item_id: int):return db.query(models.Item).filter(models.Item.id == item_id).first()# 读取多个物品 (带分页)
def get_items(db: Session, skip: int = 0, limit: int = 100):return db.query(models.Item).offset(skip).limit(limit).all()# 创建物品
def create_item(db: Session, item: schemas.ItemCreate):db_item = models.Item(name=item.name, description=item.description)db.add(db_item)db.commit()db.refresh(db_item)return db_item

第 5 步: main.py - 组装所有部分

# 文件: main.py
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session# 从我们创建的模块中导入
from . import crud, models, schemas
from .database import SessionLocal, engine# 创建数据库表 (如果它们不存在)
models.Base.metadata.create_all(bind=engine)app = FastAPI()# 依赖：为每个请求创建一个独立的数据库会话
def get_db():db = SessionLocal()try:yield dbfinally:db.close()# API 接口：创建物品
@app.post("/items/", response_model=schemas.Item)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):# 注意：这里我们不再自己处理数据库逻辑，而是调用 crud 函数return crud.create_item(db=db, item=item)# API 接口：读取多个物品
@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):items = crud.get_items(db, skip=skip, limit=limit)return items# API 接口：读取单个物品
@app.get("/items/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):db_item = crud.get_item(db, item_id=item_id)if db_item is None:raise HTTPException(status_code=404, detail="Item not found")return db_item

运行与测试：