一、先理清核心概念:FastAPI 基础
FastAPI 是一个高性能的 Python Web 框架,核心特点是自动生成接口文档、类型注解驱动、异步支持。你代码中的这些组件都是 FastAPI 为快速构建 API 提供的核心工具。
二、逐个解析组件用法
1. Query:处理 URL 查询参数
作用
专门用于定义 URL 查询参数(即 URL 中 ? 后面的参数,如 /books?search=三体),可以设置默认值、验证规则、描述等。
基础用法
from fastapi import FastAPI, Query
app = FastAPI()
# 你的代码中的用法(最基础)
@app.get("/books")
async def list_books(
# 默认值为空字符串,前端不传 search 参数时,默认就是 ""
search: str = Query("")
):
return {"search_keyword": search}
进阶用法(常用)
@app.get("/books")
async def list_books(
# 设置默认值、描述、长度限制、正则验证
search: str = Query(
"", # 默认值
title="搜索关键词", # 接口文档中显示的标题
description="按书名搜索 EPUB 书籍", # 接口文档中的描述
min_length=0, # 最小长度
max_length=100, # 最大长度
regex="^[a-zA-Z0-9\u4e00-\u9fa5]*$" # 只允许字母、数字、中文
)
):
return {"search_keyword": search}
关键说明
Query("")等价于直接写search: str = "",但用Query可以添加更多验证/描述;- 类型注解(
str)是必须的,FastAPI 会根据类型自动校验参数(比如传数字会自动转字符串); - 自动生成的接口文档(
/docs或/redoc)会显示Query设置的描述和规则。
2. UploadFile + File:处理文件上传
作用
File:标记参数是文件上传类型,配合UploadFile使用;UploadFile:FastAPI 封装的文件上传对象,提供异步/同步方法操作上传的文件,支持大文件(流式处理,不占内存)。
基础用法(你的代码中的核心场景)
from fastapi import FastAPI, UploadFile, File, HTTPException
app = FastAPI()
@app.post("/upload")
async def upload_epub(
# File(...) 表示这是必填的文件参数,UploadFile 是文件对象类型
file: UploadFile = File(...)
):
# UploadFile 的核心属性/方法
print("文件名:", file.filename) # 获取上传的文件名(如 "三体.epub")
print("文件类型:", file.content_type) # 获取文件 MIME 类型(如 "application/epub+zip")
# 保存文件(你的代码中用了 shutil.copyfileobj,也可以用异步方法)
try:
# 方式1:同步保存(你的代码用法)
with open(f"epub_storage/{file.filename}", "wb") as f:
shutil.copyfileobj(file.file, f)
# 方式2:异步保存(更推荐,适配 FastAPI 异步特性)
# contents = await file.read() # 读取文件内容(字节)
# with open(f"epub_storage/{file.filename}", "wb") as f:
# f.write(contents)
return {"status": "success", "filename": file.filename}
except Exception as e:
raise HTTPException(status_code=500, detail=f"上传失败:{str(e)}")
关键说明
File(...)中的...表示必填参数,如果前端不传文件,会直接返回 422 验证错误;UploadFile.file是一个标准的 Python 文件对象,可用于shutil.copyfileobj等操作;- 支持同时上传多个文件:
files: list[UploadFile] = File(...); - 上传完成后,FastAPI 会自动关闭文件句柄,无需手动关闭。
3. HTTPException:抛出标准化的 HTTP 异常
作用
FastAPI 提供的异常类,用于返回符合 HTTP 规范的错误响应(包含状态码和错误详情),替代普通的 raise Exception。
基础用法
from fastapi import FastAPI, HTTPException
app = FastAPI()
# 你的代码中的用法
@app.get("/download/{filename}")
async def download_epub(filename: str):
# 检查文件是否存在
if not os.path.exists(f"epub_storage/{filename}"):
# 抛出 404 异常,返回 {"detail": "文件不存在"}
raise HTTPException(
status_code=404, # HTTP 状态码(404=未找到,500=服务器错误,422=参数验证失败等)
detail="文件不存在" # 错误详情,前端可读取
)
return {"status": "success"}
常用状态码场景
| 状态码 | 含义 | 适用场景 |
|---|---|---|
| 400 | 错误的请求 | 参数格式错误 |
| 404 | 未找到 | 资源(文件/接口)不存在 |
| 403 | 禁止访问 | 权限不足 |
| 422 | 无法处理的实体 | 参数验证失败(FastAPI 自动) |
| 500 | 服务器内部错误 | 代码执行出错 |
关键说明
- 抛出
HTTPException后,FastAPI 会自动返回 JSON 格式的错误响应:{"detail": "错误信息"}; - 相比手动返回
JSONResponse(status_code=404, content={"detail": "..."}),HTTPException更简洁、符合 FastAPI 规范。
三、完整示例:组合使用所有组件
下面是一个整合所有组件的完整示例,对应你代码的核心逻辑:
import os
import shutil
from fastapi import FastAPI, UploadFile, File, HTTPException, Query
from fastapi.responses import JSONResponse
app = FastAPI(title="EPUB 上传下载示例")
UPLOAD_DIR = "epub_storage"
os.makedirs(UPLOAD_DIR, exist_ok=True)
# 1. Query:查询参数
@app.get("/books")
async def list_books(search: str = Query("", description="按书名搜索")):
books = [f for f in os.listdir(UPLOAD_DIR) if f.endswith(".epub") and search in f]
return JSONResponse(content=books)
# 2. UploadFile + File:文件上传
@app.post("/upload")
async def upload_book(file: UploadFile = File(..., description="上传EPUB文件")):
if not file.filename.endswith(".epub"):
raise HTTPException(400, detail="仅支持 EPUB 格式")
file_path = os.path.join(UPLOAD_DIR, file.filename)
try:
with open(file_path, "wb") as f:
shutil.copyfileobj(file.file, f)
return {"status": "success", "filename": file.filename}
except Exception as e:
raise HTTPException(500, detail=f"上传失败:{str(e)}")
# 3. HTTPException:异常处理
@app.get("/download/{filename}")
async def download_book(filename: str):
file_path = os.path.join(UPLOAD_DIR, filename)
if not os.path.exists(file_path):
raise HTTPException(404, detail="文件不存在")
return {"status": "ready", "path": file_path}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
总结
Query:用于定义 URL 查询参数,可设置默认值、验证规则和描述,配合类型注解实现自动校验;UploadFile + File:File标记文件参数,UploadFile封装上传文件对象,支持流式处理大文件,是 FastAPI 处理文件上传的标准方式;HTTPException:抛出标准化的 HTTP 异常,指定状态码和错误详情,替代普通异常,返回符合规范的 JSON 错误响应。
这些组件是 FastAPI 构建实用 API 的核心,结合类型注解使用,既能保证接口的规范性,又能自动生成清晰的接口文档(访问 http://localhost:8000/docs 即可查看)。
一、代码整体含义
这行代码的核心作用是:在指定目录(epub_storage)下,以“二进制写入”模式创建/打开一个与上传文件同名的文件,用于保存前端上传的 EPUB 文件内容。
二、逐部分拆解解析
1. f"epub_storage/{file.filename}":格式化文件路径
这是 Python 的 f-string 格式化字符串,作用是拼接出文件的完整保存路径:
epub_storage/:文件保存的目录(对应你代码中的UPLOAD_DIR);{file.filename}:动态替换为上传文件的原始文件名(比如三体.epub);- 最终拼接结果:比如
epub_storage/三体.epub。
2. open(...):Python 内置的文件打开函数
open() 是 Python 操作文件的基础函数,接收两个核心参数:
- 第一个参数:文件路径(上面拼接的字符串);
- 第二个参数:打开模式(这里的
"wb"是关键)。
3. "wb":文件打开模式(重点)
"wb" 是 write + binary 的缩写,即二进制写入模式,拆解说明:
| 字符 | 含义 | 关键特性 |
|---|---|---|
w | 写入模式(write) | 1. 如果文件不存在 → 自动创建 2. 如果文件已存在 → 清空原有内容(覆盖) 3. 不能读,只能写 |
b | 二进制模式(binary) | 1. 以字节(bytes)为单位处理数据 2. 适用于非文本文件(如EPUB、图片、视频、压缩包等) 3. 必须搭配 w/r/a 使用(如 wb/rb/ab) |
❗ 为什么必须用 wb 而不是 w?
- EPUB 是二进制文件(不是纯文本文件),如果用普通的
w(文本模式)打开,会对数据做编码转换(比如默认 utf-8),导致文件损坏,无法正常打开; - 只有
wb模式能原样保存二进制数据,保证上传的 EPUB 文件完整可用。
三、完整用法上下文(结合你的代码)
你的代码中这行代码是配合 with 语句使用的(资源自动管理),完整逻辑:
# 你的代码片段
with open(f"epub_storage/{file.filename}", "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
with语句:自动管理文件句柄,代码块结束后自动关闭文件(避免资源泄露);as buffer:将打开的文件对象赋值给变量buffer(缓冲区);shutil.copyfileobj(file.file, buffer):把上传文件的内容(file.file是上传文件的字节流)复制到刚创建的文件中。
四、潜在问题与优化(你的代码中已考虑,但需注意)
1. 文件名安全问题
直接使用 file.filename 有风险:
- 比如文件名包含特殊字符(如
/、\、..),可能导致路径遍历攻击(比如file.filename = "../../etc/passwd"); - 你的代码中用
safe_unicode(file.filename)处理了编码问题,还可以补充文件名过滤:# 优化:过滤非法字符 import re # 只保留字母、数字、中文、下划线、点、横线 safe_filename = re.sub(r'[^\w\u4e00-\u9fa5\.\-]', '_', file.filename) file_path = os.path.join(UPLOAD_DIR, safe_filename) with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer)
2. 目录不存在问题
如果 epub_storage 目录不存在,open() 会抛出 FileNotFoundError,你的代码中提前用 os.makedirs(UPLOAD_DIR) 创建了目录,避免了这个问题。
3. 异步优化(可选)
FastAPI 是异步框架,你代码中用的是同步的 shutil.copyfileobj,也可以用异步方式读取文件内容后写入(更适配异步场景):
# 异步写法
contents = await file.read() # 异步读取上传文件的所有字节
with open(file_path, "wb") as buffer:
buffer.write(contents) # 写入字节内容
五、对比:不同打开模式的区别(补充理解)
| 模式 | 用途 | 适用场景 |
|---|---|---|
wb | 二进制写入(覆盖) | 保存上传的二进制文件(EPUB/图片/视频) |
rb | 二进制读取 | 读取 EPUB/图片等二进制文件(比如下载接口) |
w | 文本写入(覆盖) | 保存纯文本文件(如 .txt、.json) |
a | 文本追加 | 往日志文件末尾加内容 |
总结
open(f"epub_storage/{file.filename}", "wb")核心是以二进制写入模式打开/创建文件,用于保存上传的 EPUB 二进制文件;"wb"模式是关键:w保证写入/覆盖,b保证二进制数据不被篡改;- 注意事项:需过滤文件名非法字符,提前创建保存目录,避免路径遍历攻击和文件找不到的问题。
这行代码是 Python 处理二进制文件保存的标准写法,也是文件上传功能的核心步骤之一。