【FastAPI】学习笔记

• 序言

  本文是在观看b站尚学堂官方等up的学习视频过程中的学习笔记以及多个网络帖子的参考，代码均为本人学习过程手敲，仅用于交流学习，禁止其他用途

1. 初识FastAPI

1.1. 简介

FastAPI是基于Starlette（异步Web框架）和Pydantic （数据验证库）一个现代、高性质的 Pyhon Web框架。结合了异步和类型提示的特点。

文档：

• 特点

  1. 高性能：

     • FastAPI 基于异步IO，性能接近 Node. js和GO。

     • 使用 Uvicorn (ASGI服务器），支持高并发请求。

     • 性能对比（基于 TechEmpower 等基准测试，简化为每秒请求数）：

       FastAPI：~3000 请求/秒（异步，轻量）
       Flask：~1000 请求/秒（同步，受WSGI限制）
       Django： ~800请求秒（同步，ORM和中间件开销较大

  2. 类型提示提升开发效率：

     • FastAPI 使用 Python 类型提示，通过Pydantic进行数据验证，减少手动校验代码。

     • 类型提示使代码更易读，IDE（如VSCode) 提供自动补全和错误提示。

     • 示例：定义一个带类型提示的API端点

       from fastapi import FastAPI
       from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name：str	# 字符串（默认必选）price:float		# 浮点数（默认必选）is_offer:bool = None	# 布尔值，可选@app.post("/items/")
       async def create_item(item: Item):return item.
       

  3. 自动生成API文档

     • FastAPI 内置 Swagger UI 和 ReDoc，自动生成交互式 API 文档。

  4. 异步支持

     • 支持 async/await 语法，适合高并发场景（如实时聊天、流处理）。

• 拓展

  

  • CGI 是最早的通用接口，解决服务器与动态内容生成程序的通信问题，但性能低下。
  • WSGI 针对 Python 生态优化，成为 Python Web 开发的主流标准，专注于同步 Web 应 用。
  • ASGI 是 WSGI 的升级，适应异步编程和现代 Web 需求（如 WebSocket、HTTP/2），兼容 WSGI 应用

• 问答

  FastAPI 的核心性能优势主要得益于以下哪项？

  A 、基于同步 I/O 的 Flask 框架

  B 、异步 I/O 和 Uvicorn ASGI 服务器

  C、Django 的 ORM 优化

  D、CGI 通用接口

1.2. 环境搭建

创建虚拟环境，避免环境冲突（在此之前，你需要下载anconda）

• 创建虚拟环境

  1. 创建虚拟环境

     conda create -n fastapi_env python=3.12
     

  2. 激活虚拟环境

     conda activate fastapi_env
     

     • 效果：

     

     退出虚拟环境（不必执行）

     conda deactivate
     

• 安装依赖

  1. 在虚拟环境创建FastAPI（默认安装最新的标准版）

     pip install "fastapi[standard]"
     

     这里我们指定安装版本，避免出现不必要的错误（不同的版本报错、语法可能有一些差异）

     pip install fastapi==0.115.12
     pip install uvicorn==0.34.2
     

     查看已安装的第三方包

     pip list
     

     

1.3. 第一个程序

• 第一个api程序（注意，后面的代码都会省略导入模块和启动服务相关代码）

  # --- 导入模块 ---
  from fastapi import FastAPI
  import uvicornapp = FastAPI()	# 实例化对象# 编写api接口
  @app.get("/")
  def read_root():return "Hello,World"# --- py文件方式启动服务 ---
  if __name__ == '__main__':uvicorn.run('main01:app', host='127.0.0.1', port=8000, reload=True)
  

• 启动服务

  启动上面服务的方式有如下三种：

  1. py文件方式启动服务（使用最多）

     python filename.py
     

     在 Python 文件中直接启动 Uvicorn 服务器的，这也是现阶段使用最多的启动方式，但是要像示例代码一样，设置uvicorn服务器的启动参数，如host、port等等。

     这里解释一下代码if __name__ == '__main__':

     • __name__ 是 Python 的一个内置变量，每个模块都有。
     • 当一个模块被直接运行时，它的 __name__ 会被自动设置为 '__main__'。
     • 当一个模块被导入时，它的 __name__ 会被设置为模块本身的名称（即文件名去掉 .py 后缀）。

     所以，该代码表示，if中的代码只有在第一种情况（直接运行）时才会被执行。如果这个文件是被导入的，那么这部分代码就会被跳过。

  2. 命令方式启动服务

     uvicorn filename:app_name --reload
     

     filename是文件名，app_name是FastAPI()的实例名，--reload是启动服务、并自动重新加载代码内容（热加载）

     例如：uvicorn main01:app --reload

  3. 调试方式启动服务

     fastapi dev filename.py
     

     如果提示你："To use the fastapi conmand, please install “fastapi[standard]”,安装fastapi[standard]即可(如果怕版本错误，也可以重新装回fastapi和uvicorn对应版本)

     pip install "fastapi[standard]"
     

• 查看启动效果

  因为浏览器页面的默认请求方式是GET，我们只需要，浏览器地址栏输入127.0.0.1:8000或者localhost:8000即可（不一定所有服务都是这个路径，与你自己的配置和api服务路径有关）

  

• 缺少软件包错误

  如果提示如下缺少软件包错误，选择安装缺少的软件包即可

1.4.1. 接口文档

• 访问项目生成的接口文档

  访问http://127.0.0.1:8000/docs，刚开始加载有点久，请耐心等待

  

• 接口文档调试

  

【必看】端口被占用导致的BUG

端口被旧服务（main01）占用，导致新服务（main07）的请求被旧服务拦截，清理端口占用后即可正常显示 main07 的接口结果

• 终止占用端口的进程：

  • 打开任务管理器（Windows 可按 Ctrl + Shift + Esc）。

  • 切换到「详细信息」选项卡，找到 PID 为 11048、21364 等的进程，右键选择「结束任务」。

    

当前，你也可以更改端口号来解决这个问题，但目前我还没有找到更好的解决方法来避免这个问题，用idea写Spring项目就没有碰到这个问题

2. 请求

• Request 对象

  通过注入Request 对象可获取完整的请求信息

  # Request获取请求信息
  from fastapi import FastAPI, Requestapp = FastAPI()
  @app.get('/client-info')
  async def client_info(request: Request):return {"请求URL": request.url,"请求方法": request.method,"请求IP": request.client.host,"请求参数": request.query_params,"请求头": request.headers,# "请求json": await request.json(),"请求cookies": request.cookies,# "请求form": await request.form(),# "请求files": request.files,"请求path_params": request.path_params,}
  

  • 效果：

    

2.1. 传参

2.1.1. 路径传参

python是解释型语言，对于fastapi而言，函数的顺序就是路由匹配的顺序，比如/args1/1匹配的是写在前面的path_args1()，当然，path_args2()写在前面就会匹配path_args2()

@app.get("/args1/1")
def path_args1():return {"message": "id1"}@app.get("/args2/{id}")	# 动态传入路径
def path_args2():return {"message": "id2"}

可以传递多个参数（但路径传参的多个参数一般是具有层级关系的），也可以指定参数解析类型

• 指定参数解析类型

  @app.get("/args3/{id}")
  def path_args3(id: int):return {"message": id}
  

  

• 传递多个参数（这里返回的9527应该是不带“”的）

  @app.get("/args4/{id}/{name}")
  def path_args4(id: int, name: str):return {"id": id, "name": name}
  

  

  这里应该是一个bug，我在这里还遇到一直给我报错“detail: Not Found”的错误，最后，重启一下电脑，发现文档edge浏览器被更新了，然后所有bug都没了

  

  然后返回的9527应该是又不带“”了，也是逆天，后来发现，这个bug是端口被占用导致的，解决方法我添加到了第一张

  

• 问答

  在FastAPI中，关于路径参数的说法哪个是正确的?

  A 、路径参数只能是字符串类型

  B、路径参数可以自动转换为声明的类型

  C、路径参数必须提供默认值

  D、路径参数不能包含特殊字符

2.1.1. 查询传参

声明的参数不是路径参数时，路径操作函数会把该参数自动解释为查询参数

• 查询字符串是键值对的集合，这些键值对位于 URL 的?之后，以&分隔，如

  http://127.0.0.1:8000//query1/?page=1&limit=10

  1. 简单示例

     @app.get("/query1")
     def page_limit1(page, limit):return {"page": page, "limit": limit}
     

     

     在这种情况下，如果某个参数不传递，就会报错

     

     可以看到，在项目生成的接口文档中，page和limit为必填选项

     

  2. 设非必传参数

     如下代码，设limit为None，则limit为非必传参数，再对limit进行判断，如果limit没有传递，返回信息就不包含limit

     @app.get("/query2")
     def page_limit2(page：int, limit=None):if limit:return {"page": page, "limit": limit}return {"page": page}
     

     

  3. 还可以路径传参合查询传参结合使用（在实际开发中几乎不使用这种）

     @app.get("/query3/{page}")
     def page_limit3(page, limit):return {"page": page, "limit": limit}
     

     

• 问答

  在 FastAPI 中，如果一个查询参数没有设置默认值，它的行为是什么？

  A、自动赋值为 None

  B、是必填参数，缺少时会返回错误

  C、自动赋值为 0

  D、会被忽略，不影响函数执行

2.1.3. 请求体传参

无论是路径传参还是请求体传承，都会在请求路径中暴露参数内容，而且在传递参数较多，数据量较大时，路径传参和查询传参还会使url路径过长，这时，我们可以使用请求体传参

请求体是客户端发送给 API 的数据。 发送数据使用 POST（最常用）、 PUT 、 DELETE 、 PATCH 等操作。

在简介篇我们就提到过Pydantic用于数据验证，这里我们先用Item继承BaseModel，方便后续进行参数验证

• 示例

  from pydantic import BaseModelclass Item(BaseModel):name: str   # name为字符串description: str | None = None      # description为字符串，可以不传price: float        # price为浮点数app = FastAPI()@app.post("/items/")
  async def create_item(item: Item):return item
  

  这里我们用apifox来测试

  

• 问答

  在 FastAPI 中，请求体通常以什么格式传递给 API 端点？

  A、URL 查询字符串（如 ?key=value）

  B、HTTP 头信息

  C、路径参数（如 /items/{id}）

  D、JSON 格式的请求体

2.2. 参数验证

2.2.1. 原生方式

在上面的学习中，我们已经知道了Python支持：

• 为参数指定类型，实现类型验证

• 为参数赋值None，设为非必传

其实，PythonI还支持：

• 枚举参数类型

  from typing import [e-mail]('/item1/{id}')
  def demo1(id: Union[int, str]):return {'id': id }
  

  • 效果：如图支持传入整数和字符串

    

  Optional

  与Union一样都在typing中导入，不过Optional自带允许非必传，Union[int, None]相当于Optional[int]

• 枚举参数值

  通过 Enum类限制参数为特定值。

  from enum import Enumclass ModelName(str, Enum):man = '1'woman = '0'@app.get("/models/{model_name}")
  def get_model(model_name: ModelName):return {"model_name": model_name.name}
  

  • 效果：

    

• 设置默认值（只有查询传参才支持）

  @app.get('/item2')
  def demo2(id: Union[int, str] = 123):return {'id': id }
  

  • 效果：默认为123，不传值时传入默认值

    

• 传递多个参数

  @app.get('/item3')
  def demo3(id:List):return {'id': id }
  

  • 效果

    

  这里可能会提示：You can install “python-multipart” with

  安装python-multipart即可：pip install python-multipart

• 自定义验证器(简单的验证可以直接用正则)

  这里简单解释一下：

  • Item是创建带验证的类型别名，相当于指定的可以被FastAPI识别的核心验证规则绑定器，str是验证的类型，BeforeValidator是验证之前的勾子（回调函数）

  • 而validate是绑定规则的函数

  • 相当于validate（绑定规则） => Annotated(绑定器) => Item(类型对象)

    from typing import Annotated
    from pydantic import BeforeValidatordef validate(value):if not value.startswith('P-'):raise ValueError('必须以P-开头')return value# 创建带验证的类型别名
    Item = Annotated[str, BeforeValidator(validate)]@app.get('/items1/{item_id}')
    def read_item1(item_id: Item):return {'item_id': item_id}
    

    • 效果：

      

2.2.2. Query方式

• 鼠标移动Query对象，发现Query参数还是很多的，这些参数用法不需要我们一一去记，体验一下即可，实际开发项目写多了自然就熟了

  

FastAPI 提供了强大的 Query 参数验证功能，主要通过Query类和Pydantic模型实现。

1. 基础验证

   • 类型验证：自动将参数转换为声明类型（如 int、str）。若类型不匹配（如 int 参数传入非数字），返回 422 错误。

     from fastapi import FastAPI, Query
     app = FastAPI()@app.get("/item1")
     def read_item1(q: str = Query()):return {"q": q}
     

     • 效果：

       

       1. Query的第一个参数为默认值，这里默认为None(可以不传)
       2. 如果不传参，默认传参为null

   • 必填参数：如果像设为必传，可以写为Query() 或 Query(...)，若未提供 q，返回 422 错误。

   • 长度验证：通过 min_length和 max_length限制。若 q 长度不符合要求，返回 422 错误。

     @app.get("/items2")
     def read_item2(q: str = Query(None, min_length=3, max_length=50)):return {"q": q}
     

     • 效果：

       

   • 范围验证：通过 gt（大于）、lt（小于）等限制（注意类型），若 q 数值不符合要求，返回 422 错误。

     @app.get("/item3")
     def read_item3(age: str = Query(..., gt=0, lt=100)):return {"age": age}
     

     • 效果：

       

   • 别名：仅仅是文档显示，后台传输数据不变

     @app.get("/item4")
     def read_item4(age: str = Query(..., alias= "年龄")):return {"age": age}
     

     • 效果：

       

   • 描述：通过 description参数添加文档的字段说明。

     app.get("/item5")
     def read_item5(q: str = Query(None, description="关键词", title="用户ID")):return {"q": q}
     

     • 效果：

       

   • 弃用参数：通过 deprecated=True标记弃用（一般版本更新时使用，标注弃用，但依然能用）

     @app.get("/item6")
     def read_item6(q: str = Query(None, deprecated=True)):return {"q": q}
     

     • 效果：

       

2. 正则表达式：

   • 使用 regex或者pattern参数，但推荐使用pattern，避免警告：DeprecationWarning: regex has been deprecated, please use pattern instead

   • 其中字符"^a\d{2}$"可以写为r"^a\d{2}$"，避免字符转义警告： SyntaxWarning: invalid escape sequence ‘\d’

     @app.get("/item7")
     def read_item7(q: str = Query(None, pattern = r"^a\d{2}$")):return {"q": q}
     

     • 效果：

       

3. 多值参数（列表）：使用 List类型接收多个值。

   from typing import [e-mail]("/items/")
   def read_items(q: List[str] = Query(["default"])):return {"q": q}
   

   • 效果：

   

• 问答

  在 FastAPI 中，以下哪种正则表达式验证方式是正确的？

  A、regex=“^fixedquery$”

  B、pattern=“¹+$”

  C、两者均可

  D、仅支持 regex

2.2.3. Path方式

Path 参数验证主要通过Path类和Pydantic模型实现。Path方式参数验证几乎与Query方式用法一致

 from fastapi import Path

• 问答

  在 FastAPI 中，若需限制路径参数 product_id 必须大于 1000，应如何定义？

  A、product_id: int = Path(gt=1000)

  B、product_id: int = Path(…, ge=1001)

  C、以上均可

2.2.4. Field方式

Field 是 Pydantic 提供的核心验证工具，用于为模型字段添加校验规则和元数据，使用时需要导入pydantic的Field。Field的用法与Query和Path大致相同，在Field中写入验证规则即可。

• 示例

  from pydantic import BaseModel, Fieldclass Clas1(BaseModel):name: str = Field('Tom')age: int = Field(...)@app.post('/item1')
  def item1(item: Clas1):return item
  

• 标题与描述

  class Clas2(BaseModel):name: str = Field(..., title="姓名", examples=["Tom", "Jack"])@app.post('/item2')
  def item2(item: Clas2):return item
  

  • 效果：

    

• 自定义验证器

  class Clas3(BaseModel):# 正则校验# email: str = Field(..., pattern="^[a-zA-Z0-9_-]+@[a-zA-Z0-9_-]+(\.[a-zA-Z0-9_-]+)+$")# 自定义校验email: str@field_validator('email')def validate_email(cls, value):if '@' not in value:raise ValueError('邮箱格式错误')return [e-mail]('/item3')
  def item3(item: Clas3):return item
  

  • 效果：

    

• 传递多个数据

  class Clas3(BaseModel):items: list = Field(..., min_items=1)name: str = Field(...)@app.post('/item3')
  def item3(item: Clas3):return item
  

  • 效果：

    

• 枚举

  from enum import Enumclass Gender(str, Enum):MAIN = 1WOMAN = 0class Clas4(BaseModel):gender: Gender = Field(default = Gender.WOMAN)@app.post('/item4')
  def item4():return Clas4()
  

  • 效果：

    

• 问答

  在 FastAPI 中，如何限制字符串字段长度在 3 到 20 之间？

  A、Field(min=3, max=20)

  B、Field(min_length=3, max_length=20)

  C、Field(length=(3, 20))

  D、Field(regex=“^.{3,20}$”)

2.3. 数据传输

2.3.1. 表单数据

• 安装python-multipart

  pip install python-multipart==0.0.20
  

  自 FastAPI 版本 0.113.0 起支持此功能

• 以表单方式进行传递

  • 简单示例

    from fastapi import FastAPI, [e-mail]("/item1")
    def item1(username: str = Form(...), password: str = Form(...)):return {"username": username, "password": password}
    

    • 效果：

      

  • 封装传递

    from pydantic import BaseModelclass Clas(BaseModel):username: strpassword: [e-mail]("/item2")
    def item2(item2: Clas):return item2
    

    • 效果：

      

    发现这种写法数据传递是JSON格式，具体原因是因为编译器没有识别到Clas为表单数据，这里我们可以使用在原生方式=>自定义验证器部分使用Annotated来声明Clas为表单数据

    from typing import [e-mail]("/item3")
    def item2(item2: Annotated[Clas, Form()]):return item2
    

    • 效果：

      

2.3.2. 异步处理

在 FastAPI 中，异步（async）和非异步（同步）编程方式是其核心 特性之一，它们在处理请求、性能以及并发能力上有着显著的区别。

• 异步 (async def)

  • 事件循环：异步代码运行在 Python 的异步事件循环（如 asyncio）中。事件循环负责协调多个协程（coroutines），在某个协程等待 I/O 操作（如数据库查询 或 HTTP 请求）时，事件循环可以切 换到其他协程执行。

  • 非阻塞：当一个异步函数调用 await，它会暂停执行，将控制权交回事件循环，允许其他任务运 行，直到等待的操作完成。

  • 并发性：异步模式允许单个线程处理大量并发请求，特别适合高并发场景（如 Web 服务器处理大 量客户端请求）。

• 非异步 (def)

  • 阻塞式执行：同步函数在调用时会完全占用线程，直到函数执行完成才会释放线程。
  • 线程池：在 FastAPI 中，同步函数由工作线程（worker threads）处理，Uvicorn（FastAPI 常用 的 ASGI 服务器）会将同步函数放入线程池运行。
  • 并发限制：线程池的大小限制了同步函数的并发能力。如果线程池耗尽（例如，处理大量阻塞请求），新请求将排队等待。

• 示例（不用写，运行体验即可）：

  import asyncio
  import time
  from fastapi import FastAPI, Form# 异步 endpoint：模拟并发 I/O 操作
  @app.get("/async")
  async def async_endpoint():start = time.time()# 模拟 5 次异步 I/O 操作（并发执行）tasks = [asyncio.sleep(1) for _ in range(5)]	# 创建一个异步的睡眠协程，模拟需要1秒钟的I/O操作await asyncio.gather(*tasks)	# 将多个异步任务打包，让它们并发执行end = time.time()return {"异步时长": f"{end - start:.2f}秒"}# 同步 endpoint：模拟相同的 I/O 操作
  @app.get("/sync")
  def sync_endpoint():start = time.time()# 模拟 5 次同步 I/O 操作（顺序执行）for _ in range(5):time.sleep(1)end = time.time()return {"同步时长": f"{end - start:.2f}秒"}
  

  • 效果：

    

2.3.3. 文件上传

• 单文件上传

  • bytes 类型（适合小文件）

    • 文件以二进制形式读取，适合小于 10MB 的文件

    • 内存占用高，大文件可能导致崩溃

    • 需要先创建data文件夹，避免报错：No such file or directory: ‘./data/file.jpg’

      from fastapi import FastAPI, [e-mail]("/upload1")
      def upload_file1(file: bytes = File(...)):# 文件内容直接加载到内存with open('./data/file.jpg', 'wb') as f:f.write(file)return {"msg": "上传成功"}
      

      • 效果：

        

  • UploadFile 类型（推荐）

    • 自动处理内存和磁盘存储（超过阈值存磁盘）

    • 支持文件元数据（ filename , content_type ）

    • 提供异步文件操作方法（ read() , write() ）

      最佳实践

      1. 使用 aiofiles 异步 IO 提升性能
      2. 分块读取避免内存溢出

      pip install aiofiles==24.1.0
      

      from fastapi import UploadFile
      import [e-mail]("/upload2")
      async def upload_file2(file: UploadFile):async with aiofiles.open(f'./data/{file.filename}', 'wb') as f:chunk = await file.read(1024*1024)while chunk := await file.read(1024 * 1024):  # 分块读取1MBawait f.write(chunk)return {"msg": "上传成功"}
      

      • 效果：

        

• 多文件上传

  注意：

  1. 前端需设置
  2. 每个文件独立处理，避免内存溢出

  @app.post('/upload3')
  def upload_file3(files: list[UploadFile] = File(...)):return {"count": len(files), "names": [f.filename for f in files]}
  

  • 效果：

    

• 文件验证

• 文件类型校验

  注意：HTTPException是从fastapi中导入的

  from pathlib import Path
  from fastapi import HTTPExceptionALLOW_TYPE = {'jpg', 'gif'}
  @app.post('/upload4')
  def upload_file4(file: UploadFile):type_name = Path(file.filename).suffix.lower()print('='*10, '文件格式：', type_name)if type_name not in ALLOW_TYPE:raise HTTPException(400, "不支持该文件格式")return {"msg", "文件上传成功"}
  

  • 效果：

    

• 混合表单与文件上传

  注意：

  1. 必须使用 multipart/form-data 编码
  2. 不能与 JSON 请求体（ Body ）混用

  from fastapi import Form
  @app.post("/upload5")
  def upload_file5(username: str = Form(...), avatar: UploadFile = File(...)):return {"user": username, "avatar_size": avatar.size}
  

  • 效果：

    

• 问答

  大文件上传时，以下哪项操作可能导致服务器崩溃？

  A、使用 UploadFile 分块读取

  B、用bytes直接加载10GB文件到内存

  C、限制文件大小为5MB

  D、异步保存到磁盘

3. 响应

• 常见的响应类型

  响应类型	说明
  JSON 响应	用于 RESTful API 的数据交互，占主导地位（如用户信息、订单、配置）
  列表响应	用于分页查询或批量数据返回（如商品列表、日志记录）
  文件响应	用于报表导出、文件下载（如 CSV、PDF）
  字符串响应	用于健康检查或简单状态反馈
  HTML 响应	用于管理后台或简单的 Web 页面
  重定向响应	用于认证流程或 URL 迁移
  流式响应	用于实时数据传输或大文件处理

3.1. 常用响应类型

3.1.1. JSON格式

• JSON 是 Web API 中最常见的响应格式，FastAPI 天然支持通过返 回 Python 字典或 Pydantic 模型自动序列化为 JSON 响应

  • 直接返回字典（自动序列化）

    @app.get("/item1")
    async def item1():return {"id": 1, "name": "Tom"}
    

  • 使用Pydantic模型（推荐企业实践）

    from pydantic import BaseModelclass Item(BaseModel):id: intname: strtags: list[str] = []@app.get("/item2", response_model = Item)
    async def item2():return Item(id = 2, name = "Tom")
    

    • 效果：

      

      还可以设置response_model_exclude_unset= True，这样只有设置了值才会返回，如tags为空列表，就不返回tags

  • 自定义响应类型：

    就是不同的企业或项目组有不同的返回规范

    from typing import Union, TypeVar, Generic# 自定义泛型模型
    T = TypeVar("T")# T为定义泛型类型的变量
    class SuccessResponse(BaseModel, Generic[T]):status: str = "success"data: Tclass ErrorResponse(BaseModel):status: str = "error"message: strcode: [e-mail]("/item3/{id}", response_model=Union[SuccessResponse[Item], ErrorResponse])
    async def item3(id: int):if id == 1:# 定义要返回的数据item = Item(id=3, name="Tom", tags=["red", "black"])return SuccessResponse[Item](data=item)else:return ErrorResponse(message="Item没有找到", code=404)
    

    • 效果：

      

• 问答：

  1. Web API 中最常见的响应格式是什么？

     A、HTML 响应

     B、JSON 响应

     C、文件响应

     D、流式响应

  2. 在 FastAPI 中，哪个功能被推荐用于企业级应用以确保响应数据 验证和自动生成 OpenAPI 文档？

     A、response_model

     B、response_model_exclude_unset

     C、Union[SuccessResponse, ErrorResponse]

     D、Pydantic BaseModel

3.1.2. 列表格式

列表响应是指 API 返回一组数据的响应，通常以 JSON 格式返回一个数组或包含数组的对象。常见场景包括：

1. 分页查询：返回数据的一部分（如每页 10 条记录），避免一次 性加载所有数据。
2. 批量数据返回：返回符合条件的全部或部分数据（如所有订单、 日志）。
3. 列表响应通常包含：
   • 数据列表：核心数据（例如商品列表、日志记录）
   • 分页元数据：如总记录数、当前页码、每页记录数、总页数等
   • 状态信息：如请求状态（成功或失败）。
   • 过滤/排序信息：描述当前返回的数据是如何过滤或排序的

• 列表响应：

  # 定义商品模型
  class Item(BaseModel):id: intname: strprice: floatcategory: strdb = [Item(id=i, name=f"Item {i}", price=i, category="item" if i % 2 == 0 else "-item-") for i in range(1, 101)]@app.get("/item1")
  async def item1():return db
  

  • 效果：

    

• 数据过滤与分页

  from typing import List, Optional
  from fastapi import [e-mail]("/item2")
  async def item2(# 对数据分页page: int = Query(1, ge=1, description='页码'),page_size: int = Query(10, ge=1, description='页数'),category: Optional[str] = Query(None, description='分类')
  ):# 对数据进行过滤sublist = dbif category:sublist = [i for i in db if i.category == category]total = len(sublist)    # 过滤后的数据条数total_pages = (total + page_size - 1) // page_size  # 过滤后的页数start = (page - 1) * page_size  # 要返回的数据在sublist的开始索引end = start + page_size # 要返回的数据在sublist的结束索引return sublist[start: end]
  

  • 效果：

    

• 分页列表封装

  • 这里老师的代码写得有点不合理，在实际开发中，返回的total应该是category字段条件过滤后的总条数，而不是所有数据的总条数

  • 另外需要注意的是，过滤操作一般是查询数据库时SQL语句来实现，避免sublist = db浪费内存

    class Pagination(BaseModel):total: inttotal_pages: intpage: intpage_size: intclass ListResponse(BaseModel):data: List[Item]pagination: Paginationcode: str = "200"@app.get("/item3")
    async def item3(# 对数据分页page: int = Query(1, ge=1, description='页码'),page_size: int = Query(10, ge=1, description='页数'),category: Optional[str] = Query(None, description='分类')
    ):# 对数据进行过滤sublist = dbif category:sublist = [i for i in db if i.category == category]return ListResponse(data=sublist[(page - 1) * page_size: page * page_size],pagination=Pagination(total=len(sublist),     # 过滤后的数据条数total_pages=(len(sublist) + page_size - 1) // page_size,   # 过滤后的页数page=page,page_size=page_size))
    

    • 效果：

      

• 问答：

  列表响应的主要目的是什么？

  A、返回单个数据记录的详细信息

  B、返回一组数据的集合，通常以数组形式

  C、更新服务器上的数据

  D、删除数据库中的记录

3.1.3. 文件格式

文件响应是指 API 在响应客户端请求时，返回一个文件内容的 HTTP 响应，通常包含文件的二进制数据或文本数据

• 内容类型：通过 Content-Type 头指定文件的 MIME 类型，如 application/pdf（PDF 文件）、 text/csv（CSV 文件）、application/vnd.ms-excel（Excel 文件）

• 常用的MIME类型

  • 文本类型

    MIME类型	描述	文件扩展名
    text/plain	纯文本	.txt
    text/html	HTML文档	.html, .htm
    text/css	CSS样式表	.css
    text/javascript	JavaScript代码	.js
    text/csv	CSV数据	.csv
    text/xml	XML数据	.xml
    text/markdown	Markdown文档	.md

  • 图像类型

    MIME类型	描述	文件扩展名
    image/jpeg	JPEG图像	.jpg, .jpeg
    image/png	PNG图像	.png
    image/gif	GIF图像	.gif
    image/svg+xml	SVG矢量图像	.svg
    image/webp	WebP图像	.webp
    image/x-icon	图标	.ico

  • 应用类型

    MIME类型	描述	文件扩展名
    application/json	JSON数据	.json
    application/xml	XML数据	.xml
    application/pdf	PDF文档	.pdf
    application/zip	ZIP压缩文件	.zip
    application/gzip	GZIP压缩文件	.gz
    application/octet-stream	二进制数据	(任意)
    application/x-www-form-urlencoded	表单数据	-

  • 音频类型

    MIME类型	描述	文件扩展名
    application/json	JSON数据	.json
    application/xml	XML数据	.xml
    application/pdf	PDF文档	.pdf
    application/zip	ZIP压缩文件	.zip
    application/gzip	GZIP压缩文件	.gz
    application/octet-stream	二进制数据	(任意)
    application/x-www-form-urlencoded	表单数据	-

  • 视频类型

    MIME类型	描述	文件扩展名
    video/mp4	MP4视频	.mp4
    video/webm	WebM视频	.webm
    video/ogg	OGG视频	.ogv
    video/x-msvideo	AVI视频	.avi

  • 文档类型

    MIME类型	描述	文件扩展名
    application/msword	Word文档	.doc
    application/vnd.openxmlformats-officedocument.wordprocessingml.document	Word文档 (docx)	.docx
    application/vnd.ms-excel	Excel文档	.xls
    application/vnd.openxmlformats-officedocument.spreadsheetml.sheet	Excel文档 (xlsx)	.xlsx
    application/vnd.ms-powerpoint	PowerPoint文档	.ppt
    application/vnd.openxmlformats-officedocument.presentationml.presentation	PowerPoint文档 (pptx)	.pptx

  • 其他类型

    MIME类型	描述	文件扩展名
    multipart/form-data	多部分表单数据	-
    message/rfc822	电子邮件消息	.eml

• 文件内容：响应的 body 包含文件的实际数据，可能是二进制（如 PDF、图片）或文本（如 CSV、 JSON 文件）

  • 文本信息（Response）

    from fastapi.responses import Response
    @app.get('/item1')
    async def item1():info = b'Hello World'	# b代表bytes类型return Response(content=info,	# 内容media_type='text/plain',	# 媒体类型headers={'Content-Disposition': 'attachment;filename="file.txt"'}	# attachment为直接下载（附件）)
    

    • 效果：

      

  • pdf文件（FileResponse）

    from fastapi.responses import FileResponse
    @app.get('/item2')
    async def item2():path = './files/hello.pdf'return FileResponse(content=path,	# 内容media_type='application/pdf',	# 媒体类型headers={'Content-Disposition': 'attachment;filename="file.pdf"'}	# attachment为直接下载（附件）)
    

    • 效果：

      

  • mp4文件（StreamingResponse）

    from fastapi.responses import StreamingResponse
    def generate_chunks(file_path: str, chunk_size: int = 1024*1024*10):with open(file_path, 'rb') as f:while chunk := f.read(chunk_size):yield [e-mail]('/item3')
    async def item3():path = './files/demo.mp4'return StreamingResponse(content = generate_chunks(path),media_type='video.mp4',	# 媒体类型headers={'Content-Disposition': 'attachment;filename="file.mp4"'}	# attachment为直接下载（附件）)
    

3.2. 其他响应类型

• 字符串格式

  @app.get('/item1')
  async def item1():return "Hello"
  

• HTML格式（HTMLResponse）

  from fastapi.responses import HTMLResponse# response_class=HTMLResponse为自动解析html
  @app.get('/item2', response_class=HTMLResponse)
  async def item2():return "<html><h1>Hello</h1></html>"
  

  • 效果：

    

• 重定向响应（RedirectResponse）

  重定向后，跳转页面到重定向后的页面，下面示例在重定向的同时将参数也传递过去了

  from fastapi.responses import [e-mail]('/item4', response_class=HTMLResponse)
  async def item4(name: str):return f"<html><h1>{name}</h1></html>"@app.get('/item5')
  async def item5():return RedirectResponse(url='/item4?name=Tom')
  

• 静态文件格式

  可以直接将html文件作为静态文件挂载到fastapi实例上来访问

  • html文件

    <!DOCTYPE html>
    <html lang="en">
    <head><meta charset="UTF-8"><title>hello</title>
    </head>
    <body><h1>hello</h1>
    </body>
    </html>
    

  • 挂载代码

    from fastapi.staticfiles import StaticFiles
    app.mount('/page', StaticFiles(directory='files', html= True))
    

  • 效果：

    

4. ORM

ORM（Object-Relational Mapping，对象关系映射）是一种编程技术，用于在面向对象编程语言和关系型数据库之间建立映射。它允许开发者通过操作对象的方式来与数据库进行交互，而无需直接编写复杂的 SQL 语句。

• 主要特点：

  • 对象与数据库表的映射：ORM 将数据库中的表映射为编程语言中的类，每一行数据对应一个对象，表的列对应对象的属性
  • 简化数据库操作：开发者可以使用面向对象的方法（如创建、查询、更新、删除对象）来操作数据 库，而无需手动编写 SQL
  • 跨数据库兼容：ORM 通常支持多种数据库（如 MySQL、PostgreSQL、SQLite），通过统一的接 口减少数据库切换的成本
  • 提高开发效率：通过自动化 SQL 生成和查询优化，减少重复代码，提升开发速度

• ORM工具介绍：

  • SQLAlchemy（同步/异步）：≈80% 企业项目首选，功能完备、社区成熟，支持复杂查询和事务 管理。
  • Tortoise（异步）：语法类似 Django ORM，适合异步优先项目，集成简便
  • GINO（异步）：轻量级，基于 SQLAlchemy Core 的异步扩展，适合高性能 API

• 问答：

  1. 什么是 ORM 的主要功能？

     A、直接编写 SQL 语句来操作数据库

     B、将数据库表映射为编程语言中的类，实现对象操作数据库

     C、优化数据库的存储结构

     D、替换数据库管理系统

  2. 以下哪项不是 ORM 的优点？

     A、提高代码可读性和维护性

     B、减少直接 SQL 操作带来的错误

     C、性能总是优于原生 SQL

     D、支持复杂查询和关系

4.1. Tortoise-ORM

4.1.1. 配置

• 环境配置

  pip install tortoise-orm==0.25.0 aerich==0.9.0 aiomysql==0.2.0 tomlkit==0.13.2
  

• 数据库连接配置

  1. connections：定义数据库连接字符串

     • SQLite：sqlite://db.sqlite3 （开发环境，文件存储在项目根目录）

     • PostgreSQL： postgres://user:password@host:port/dbname （生产环境）

     • MySQL： mysql://user:password@host:port/dbname （生产环境）

  2. apps：定义应用模块，models 列表包含模型文件路径和 Aerich 的迁移模型

  3. db_pool：连接池参数，优化数据库连接管理，适合高并发场景

  4. use_tz 和 timezone：控制时间字段的时区行为（生产环境建 议启用）

     from typing import Dict# Tortoise-ORM 配置
     TORTOISE_ORM: Dict = {"connections": {# 开发环境使用 SQLite（基于文件，无需服务器）# "default": "sqlite://db.sqlite3",# 生产环境示例：PostgreSQL# "default":"postgres://user:password@localhost:5432/dbname",# 生产环境示例：MySQLdefault":"mysql://root:123456@localhost/fastapi"},"apps": {"models": {"models": ["aerich.models"],	# 模型模块和 Aerich 迁移模型"default_connection": "default",}},# 连接池配置（推荐）"use_tz": False,  # 是否使用时区"timezone": "UTC",  # 默认时区\"db_pool": {"max_size": 10,  # 最大连接数"min_size": 1,# 最小连接数"idle_timeout": 30  # 空闲连接超时（秒）}
     }from tortoise.contrib.fastapi import register_tortoise
     register_tortoise(app, # 关联FastAPI实例config=TORTOISE_ORM, # 数据库配置generate_schemas=True, # 是否自动生成数据库表结构add_exception_handlers=True   # 是否添加异常处理)@app.get("/")
     async def root():return {"message": "Welcome to FastAPI with Tortoise-ORM!"}
     

• 问答：

  1. 关于Tortoise-ORM 的说法正确的是？

     A、它只支持同步操作

     B、它与 FastAPI 无缝集成并支持异步操作

     C、它不支持复杂关系

     D、它需要手动生成表结构

  2. 在配置 Tortoise-ORM 时，以下哪项是正确的数据库连接字符串示例？

     A、mysql://user:password@localhost:3306/dbname

     B、http://localhost:5432/dbname

     C、sqlite://user:[e-mail]3

     D、postgres://dbname@localhost:5432

4.1.2. Aerich 迁移工具

Aerich 是 Tortoise-ORM 的数据库迁移工具，用于管理数据库结构的变更

• 示例模型文件（model19.py）：

  from tortoise.models import Model
  from tortoise.fields import CharField,DatetimeField,BooleanFieldclass User(Model):id = CharField(max_length=36, pk=True) # 主键，UUID 字符串username = CharField(max_length=50, unique=True)  # 用户名，唯一email = CharField(max_length=255, unique=True)  # 邮箱，唯一is_active = BooleanField(default=True) # 是否激活created_at = DatetimeField(auto_now_add=True)  # 创建时间updated_at = DatetimeField(auto_now=True)  # 更新时间class Meta:table = "users"  # 自定义表名ordering = ["-created_at"]  # 默认按创建时间降序排序def __str__(self):return self.username
  

  在迁移之前，要先在main19.py中关联示例模型文件model19.py模型文件

  "models": {"models": ["model19", "aerich.models"],    # 模型模块和 Aerich 迁移模型"default_connection": "default",
  }
  

• Aerich初始化

  在项目根目录运行以下命令(这里main是不带后缀的文件名，比如我是main19)

  aerich init -t main.TORTOISE_ORM
  

  这将生成

  • pyproject.toml：Aerich 配置文件，指定迁移配置。

  • migrations/：迁移文件目录，存放生成的 .sql 文件。

    aerich init-db
    

    • 效果：

      

      生成了数据库表

      

• 生成和应用迁移

  • 生成迁移文件： 当模型发生变更时，运行以下命令生成迁移文件

    aerich migrate --name "注释"
    

  • 应用迁移： 运行以下命令将迁移应用到数据库

    aerich upgrade+ **验证迁移**： 检查迁移历史```python
    aerich history
    

  • 回滚迁移：回退到指定版本

    aerich downgrade
    

• 问答

  Aerich 的主要功能是什么？

  A、提供异步 ORM 的模型定义

  B、自动生成 FastAPI 路由

  C、优化数据库连接池

  D、管理数据库结构的变更

4.1.3. 模型定义

• 使用方法：

  • 继承 Model：Tortoise-ORM 的模型通过继承 tortoise.models.Model 类定义， 每个模型对应数据库中的一张表。

  • 字段定义：字段定义使用 Tortoise-ORM 提供的字段类（如 CharField、 BooleanField），结合字段参数来指定约束和行为。

  • 常用的字段类型

    字符类型	数据库类型	描述	常用参数
    CharField	VARCHAR	字符串	max _length （必填）、 unique=True 、 index=True
    TextField	TEXT	长文本	null=True（允许为空）
    IntField / BigIntField	INTEGER/BIGINT	整数	pk=True（主键）、 default=0
    FloatField/ DecimalField	DOUBLE/REAL/ DECIMAL	浮点数/高精度小数	max_digits=10 , decimal_places=2 （最大10位，含2位小数）
    BooleanField	BOOLEAN	布尔值	default=True
    DateField / DatetimeField	DATE/ DATETIME/TIMESTAMP	日期/日期时间	auto_now_add=True （仅首次保存记录时间）
    JSONField	JSON/JSONB	存储字典或列表	encoder （自定义编码器）、 decoder （自定义解码器）
    UUIDField	UUID/CHAR(36)	唯一标识符	若为主键默认生成 UUID4
    BinaryField	BLOB	二进制数据	不支持过滤或更新操作

  • 字段参数

    参数	说明
    max_length	字符串字段的最大长度（CharField 必填）
    null	是否允许字段为空（null=True 表示数据库允许 NULL）
    default	字段默认值（如 default=0、default=True）
    unique	是否唯一（unique=True 确保字段值在表中唯一）
    index	是否创建索引（index=True 提高查询性能）
    description	字段描述（用于文档或数据库注释）
    pk	是否为主键（pk=True 表示该字段是主键）
    validators	：自定义验证函数

    • 示例：

      from tortoise.validators import Validator
      def validate_credit(value):if value < 0:raise ValueError("信用值不能为负数")class User(Model):credit = fields.IntField(validators = [validate_credit])  # 自定义验证函数 
      

  • Meta 类：用于定义模型的元数据，如表名、排序规则等，常用属性包括：

    • table：自定义数据库表名（如 table=“users”）

    • unique_together：定义联合唯一约束（如 unique_together=[(“field1”, “field2”)]）

    • indexes：定义索引（如 indexes=[(“field1”, “field2”)]）

    • ordering：默认排序规则（如 ordering=[“-created_at”, “username”]）

    • 示例：

      class Event(Model):name = CharField(max_length=100)location = CharField(max_length=200)date = DateField()class Meta:table = "events"unique_together = [("name", "date")] # 名称和日期联合唯一indexes = [("location",)]  # 为 location 字段创建索引ordering = ["date"]  # 按日期升序排序
      

    基本模型示例：

    from tortoise.fields import CharField, BooleanField, DatetimeField
    from tortoise.models import Modelclass User(Model):id = CharField(max_length=36, pk=True)username = CharField(max_length=50, unique=True)age = CharField(max_length=3)email = CharField(max_length=255, unique=True)is_active = BooleanField(default=True) created_at = DatetimeField(auto_now_add=True)	# 创建时间class Meta:table = "users"     # 自定义表名unique_together = ["username", "email"]     # 联合唯一ordering = ["-created_at"]      # 默认排序, -降序, +升序
    

• 问答：

  1. 在 Tortoise-ORM 中，定义一个模型时必须继承哪个类？

     A、tortoise.Model

     B、tortoise.models.Model

     C、tortoise.BaseModel

     D、tortoise.orm.Model

  2. 以下哪个字段类型适合存储高精度小数（如货币金额）？

  A、DecimalField

  B、FloatField

  C、IntField

  D、CharField

4.1.4. CRUD

• 示例代码（包含基本信息的学生模型，用于演示单表的增删改查操作）：

  from tortoise import fields, modelsclass Student(models.Model):id = fields.IntField(pk=True, description="学生ID，主键")name = fields.CharField(max_length=50, description="学生姓名")age = fields.IntField(null=True, description="学生年龄，可为空")email = fields.CharField(max_length=100, unique=True, null=True, description="学生邮箱，唯一")class Meta:table = "students"def __str__(self):return f"Student: {self.name}, Age: {self.age}, Email: {self.email}"
  

  • 增（Create）：create_student

    1. 直接脚本测试

       from model21 import Student
       from tortoise import Tortoise, run_asyncasync def init():await Tortoise.init(db_url="mysql://root:123456@localhost/fastapi",modules={"models": ["model21"]})await Tortoise.generate_schemas()   # 生成数据库表结构(初始化)async def create_student(name: str, age: int, email: str) -> Student:try:stu = await Student.create(name=name, age=age, email=email)return stuexcept Exception as e:print("创建失败---", e)# 1. 直接脚本测试
       async def main():await init()stu = await create_student("Tom", 18, "[e-mail]")     # 创建数据print(f"脚本创建成功, id:{stu.id}, name:{stu.name}, email: {stu.email}")await Tortoise.close_connections()      # 关闭数据库连接if __name__ == '__main__':run_async(main())   # 异步调用

       • 效果：

         

    2. 直接应用到接口（在main21中写如下面代码并启动main21）

       @app.get("/add")
       async def add():stu = await create_student("jack", 18, "[e-mail]")return stu# 生成一个传递请求体参数创建学生数据的路由
       @app.post("/add")
       async def add(name: str, age: int, email: str):stu = await create_student(name, age, email)return stu
       

       • 效果：

         

  • 删（Delete）：delete_student（仅展示关键代码）

    async def delete_student(id: int):try:stu = await Student.get(id=id)await stu.delete()print(f"删除成功, id:{stu.id}")except Exception as e:print("删除失败---", e)async def main():await init()await delete_student(3)
    

    • 效果：

      

  • 改（Update）：update_student（仅展示关键代码）

    async def update_student(id: int=None, name: str=None, age: int=None, email: str=None):try:stu = await Student.get(id=id)stu.name = namestu.age = agestu.email = emailawait stu.save()print(f"更新成功, id:{stu.id}")except Exception as e:print("更新失败---", e)# 1. 直接脚本测试
    async def main():await init()await update_student(1, "Tom", 18, "[e-mail]")
    

    • 效果：

      

  • 查（Read）（仅展示关键代码）

    1. 单条数据

       async def query_student(id: int):try:stu = await Student.get(id=id)print(f"查询成功, id:{stu.id}, name:{stu.name}, email: {stu.email}")except Exception as e:print("查询失败---", e)# 1. 直接脚本测试
       async def main():await init()await query_student(1)
       

       • 效果：

         

    2. 多条数据 + 精确匹配

       async def query_student1(id: int) -> list[Student]:stus = await Student.filter(id=id)return stus
       

    3. 多条数据 + 模糊匹配

       async def query_student2(name: str) -> list[Student]:stus = await Student.filter(name__icontains=name)return stus
       

    4. 多条数据 + 全部查询

       async def query_student3() -> list[Student]:stus = await Student.all()return stus
       

4.1.5. 关联关系

• 我们将模拟一个学校管理系统，包含以下实体：

  • 学生（Student）：每个学生有唯一的个人信息档案（1对1）。
  • 成绩（Grade）：一个学生可以有多份成绩记录（1对多）。
  • 课程（Course）：学生和课程之间是多对多关系（通过成绩表关联）。

  示例代码：

  from tortoise import fields, modelsclass Student(models.Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=50)  # 1对1关系：学生和学生档案profile = fields.OneToOneField("models.StudentProfile", on_delete=fields.CASCADE, related_name="student")# 1.对多关系：学生和成绩（通过反向关系访问）grades = fields.ReverseRelation["Grade"]class Meta:table = "students"def __str__(self):return f"Student: {self.name}"class StudentProfile(models.Model):id = fields.IntField(pk=True)student = fields.OneToOneField("models.Student", on_delete=fields.CASCADE, related_name="profile")address = fields.CharField(max_length=100, null=True)phone = fields.CharField(max_length=20, null=True)class Meta:table = "student_profiles"def __str__(self):return f"Profile for {self.student.name}: {self.address}, {self.phone}"class Course(models.Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=50)# 多对多关系：通过Grade表关联学生students = fields.ManyToManyField("models.Student", through="grades", related_name="courses")class Meta:table = "courses"def __str__(self):return f"Course: {self.name}"class Grade(models.Model):id = fields.IntField(pk=True)student = fields.ForeignKeyField("models.Student", related_name="grades", on_delete=fields.CASCADE)course = fields.ForeignKeyField("models.Course", related_name="grades", on_delete=fields.CASCADE)score = fields.FloatField()class Meta:table = "grades"unique_together = ("student", "course")  # 确保学生和课程组合唯一def __str__(self):return f"{self.student.name} - {self.course.name}: {self.score}"
  

• 一对一

  # 1对1
  from tortoise import fields, models, Tortoise, run_asyncclass Student(models.Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=50)profile = fields.OneToOneField("models.StudentProfile",    # 关联模型related_name="student",     # 反向关联名称(StudentProfile反向查Student)on_delete=fields.CASCADE,   # 删除级联(Student数据删除，StudentProfile中对应数据也被删除))class StudentProfile(models.Model):id = fields.IntField(pk=True)address = fields.CharField(max_length=255)phone = fields.CharField(max_length=20)async def init():await Tortoise.init(db_url="mysql://root:123456@localhost/fastapi2",modules={"models": ["model22"]})await Tortoise.generate_schemas()if __name__ == '__main__':run_async(init())
  

  • 效果：

    

• 一对多

  一对多把关联关系创建在多的那边

  # 一对多
  from tortoise import fields, models, Tortoise, run_asyncclass Student(models.Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=50)class Grade(models.Model):id = fields.IntField(pk=True)score = fields.FloatField()student = fields.ForeignKeyField("models.Student",related_name="grades",on_delete=fields.CASCADE,)async def init():await Tortoise.init(db_url="mysql://root:123456@localhost/fastapi3",modules={"models": ["model23"]})await Tortoise.generate_schemas()if __name__ == '__main__':run_async(init())
  

  • 效果：

    

• 多对多

  多对多需要创建中间表来关联

  # 多对多
  from tortoise import fields, models, Tortoise, run_asyncclass Student(models.Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=50)class Course(models.Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=50)students = fields.ManyToManyField("models.Student",through="student_course",   # 中间表名称related_name="courses")async def init():await Tortoise.init(db_url="mysql://root:123456@localhost/fastapi4",modules={"models": ["model24"]})await Tortoise.generate_schemas()if __name__ == '__main__':run_async(init())
  

  • 效果：

    

  但是，中间表有可能会有重复数据，这在实际业务中可能不被允许，我们可以通过自己手动中间表，并用Meta约束唯一来解决这个问题

  class Course(models.Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=50)class StudentCourse(models.Model):students = fields.ForeignKeyField("models.Student", related_name="courses")courses = fields.ForeignKeyField("models.Course", related_name="students")class Meta:unique_together = ("students", "courses")

• 问答：

  1. 在一对一关系中，两个表之间的关联特点是什么？

  A、一个表中的一条记录可以对应另一个表中的多条记录

  B、一个表中的一条记录只对应另一个表中的一条记录

  C、两个表中的记录可以任意组合，无限制

  D、一个表中的多条记录对应另一个表中的一条记录

  2. 以下哪种场景适合使用多对多关系？

     A、一个用户只能有一个邮箱地址，一个邮箱地址只能属于一个用 户

     B、一个部门可以有多个员工，一个员工只能属于一个部门

     C、一个学生可以选修多门课程，一门课程可以被多个学生选修

     D、一个订单只能属于一个客户，一个客户可以有多个订单

4.1.6. 关联关系操作

• 增

  • 设置profile_id可以为空

    class Student(models.Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=50)profile = fields.OneToOneField("models.StudentProfile",related_name="student",on_delete=fields.CASCADE,null = True,    # 让profile_id可以为空)
    

  • 增加数据

    • 一对一

      from tortoise import Tortoise, run_async
      from model24 import Student, Course, StudentCourse, Grade, StudentProfileasync def init():await Tortoise.init(db_url="mysql://root:123456@localhost/fastapi4",modules={"models": ["model24"]})await Tortoise.generate_schemas()async def create_data():await init()# ----- 一对一 -----# stu1 = await Student.create(name="Tom")# pro1 = await StudentProfile.create(address="北京", phone="12345678901")# stu1.profile = pro1# await stu1.save()# 在创建数据项时直接绑定个人信息pro2 = await StudentProfile.create(address="北京", phone="12345678901")stu2 = await Student.create(name="Tom", profile=pro2)await stu2.save()if __name__ == '__main__':run_async(create_data())
      

      • 效果：

        

    • 一对多

      stu3 = await Student.create(name="Jack")
      await Grade.create(score=100, student=stu3)
      

      • 效果：

        

    • 多对多

      stu4 = await Student.create(name="Bob")
      course4 = await Course.create(name="Python")
      await StudentCourse.create(students=stu4, courses=course4)
      

      • 效果：

        

• 删

  async def delete_data():await init()grade = await Grade.get(id=9)await grade.delete()
  

• 改

  async def update_data():await init()# ----- 一对一 -----pro1 = await StudentProfile.create(address="中国-北京", phone="12345678901")stu1 = await Student.get(id=1, profile=pro1)await stu1.save()# ----- 一对多 -----stu2 = await Student.get(id=1)gradel1 = await Grade.get(id=1, student=stu2)await gradel1.save()# 通过类进行更新await Grade.filter(id = 1).update(student=stu2)# ----- 多对多 -----stu3 = await Student.get(id=1)course3 = await Course.get(id=1)stu4 = Student.create(name="Jim")course4 = Course.create(name="Java")await StudentCourse.filter(students=stu3, courses=course3).update(students=stu4, courses=course4)
  

• 查

  async def query_data():await init()# ----- 一对一 -----stu1 = await Student.get(id=1)print(stu1)pro = await stu1.profileprint(pro.address)stu2 = await Student.get(id=1).prefetch_related("profile")print(stu2.profile.address)print(stu2.profile.phone)# ----- 一对多 -----stu3 = await Student.get(id=9).prefetch_related("grades")for grade in stu3.grades:print(grade.score)
  

5. 其他

5.1. 中间件

中间件是位于客户端和应用程序核心逻辑之间的软件层，常用于：

• 拦截请求和响应

• 在请求到达路由处理程序之前进行处理

• 在响应返回给客户端之前进行处理

  

• 示例

  from fastapi.responses import [e-mail]('http')
  # request为请求对象，call_next为下一个处理函数
  async def middleware1(request, call_next):print('处理业务之前...')print(request.method, request.url)# 运行部分url通过# if request.url.path == '/item1':# 	return Response(content="没有权限访问该接口")response = await call_next(request)response.headers['X-token'] = '123456'print('处理业务之后...')return [e-mail]('/item1')
  async def item1():print('=== 处理中 ===')return '处理完毕'
  

  • 效果：

    

  中间件还要一种写法(偏底层，了解即可)

  class LogMiddleware:def __init__(self, app):self.app = appasync def __call__(self, scope, receive, send):print('处理业务之前...')await self.app(scope, receive, send)print('处理业务之后...')
  

• 中间件的嵌套

  关于为什么上面的中间件后执行，参考上面的中间件运行图，可以理解为越后面的中间件越靠近浏览器（用户端）

  @app.middleware('http')
  async def middleware2(request, call_next):print('中间件2：处理业务之前...')response = await call_next(request)print('中间件2：处理业务之后...')return [e-mail]('http')
  async def middleware1(request, call_next):print('中间件1：处理业务之前...')response = await call_next(request)print('中间件1：处理业务之后...')return [e-mail]('/item1')
  async def item1():print('=== 处理中 ===')return '处理完毕'
  

  • 效果：

    

• 问答：

  在 FastAPI 中，如何正确添加一个自定义中间件？

  A、使用 @app.middleware(“http”) 装饰器

  B、使用 @app.route() 装饰器

  C、直接在路由函数中定义中间件逻辑

  D、使用 @app.exception_handler 装饰器

5.2. 跨域共享

• 同源策略（SOP）：浏览器的一种安全机制，限制了一个源（origin）的网 页如何与另一个源的资源进行交互。

  • 源（origin）由协议（如HTTP/HTTPS）、域名（如example.com）和端口（如80或443）组成。 (例如， https://example.com:443 和 http://example.com:80 是不同源，因为协议和端口不同)。
  • 同源策略防止恶意网站通过脚本（如JavaScript）未经授权访问其他 网站的数据，例如窃取用户的敏感信息

  但是：现代Web应用经常需要跨源请求！

• CORS（Cross-Origin Resource Sharing，跨源资源共享）：一种 基于HTTP的机制，它允许服务器指示哪些其他源（域名、协议或端 口）可以访问其资源，从而绕过浏览器的同源策略（Same-Origin Policy，SOP）限制

跨域共享实现：

• 测试代码（text.html）：

  <!DOCTYPE html><html><head><title>CORS Test</title></head><body><h1>CORS 测试</h1><button onclick="testCors()">测试CORS</button><p id="result">这块儿显示响应</p><script>async function testCors() {try {const response = await fetch('http://127.0.0.1:8000/info', {method: 'GET',headers: {'Content-Type': 'application/json'}});const data = await response.json();document.getElementById('result').textContent = 'Success: ' + JSON.stringify(data);}catch (error) {document.getElementById('result').textContent = 'Error: ' + error.message;}}</script></body>
  </html>
  

  • 效果：

    

  跨域共享实现（了解即可，后面可以用工具方法跨域）：

  @app.middleware('http')
  async def add_cors_headers(request, call_next):response = await call_next(request)response.headers['Access-Control-Allow-Origin'] = '*'	# 允许所有源访问return  [e-mail]('/info')
  async def info():return '内容获取成功'
  

  • 效果：

    

  如上图所示，还是被拒绝了

  修改后的跨域共享实现（了解即可，后面可以用工具方法跨域）：

  from fastapi.responses import [e-mail]('http')
  async def add_cors_headers(request, call_next):# 处理OPTIONS请求if request.method == 'OPTIONS':headers = {'Access-Control-Allow-Origin': '*',		# 允许所有源访问'Access-Control-Allow-Methods': 'POST, GET, OPTIONS, PUT, DELETE',		# 允许的请求方法'Access-Control-Allow-Headers': 'Content-Type, Authorization',		# 允许的请求头}return Response(status_code=200, headers=headers)response = await call_next(request)response.headers['Access-Control-Allow-Origin'] = '*'	# 允许所有源访问return  [e-mail]('/info')
  async def info():return '内容获取成功'
  

  • 效果：

    

• 利用工具方法实现跨域

  from fastapi.middleware.cors import CORSMiddleware
  app.add_middleware(CORSMiddleware,allow_origins=['*'],	# 允许的源访问，如[http://127.0.0.1:8080]allow_methods=['*'],	# 允许的请求方法allow_headers=['*'],	# 允许的请求头allow_credentials=True,	# 允许携带cookie
  )
  

• 问答：

  CORS（跨源资源共享）的主要目的是什么？

  A、完全禁止所有跨源请求以确保安全性

  B、允许服务器控制哪些跨源请求可以访问其资源

  C、强制所有Web请求使用HTTPS协议

  D、提高浏览器的页面加载速度

5.3. APIRouter

• APIRouter 核心作用

  • 模块化架构：将大型应用拆分为独立功能模块
  • 路由分组：统一管理相关端点
  • 组织优化：解耦业务逻辑，提升可维护性

• 简单示例

  from fastapi import APIRouter
  router = APIRouter() # 创建路由器goods_router = APIRouter(tags=["商品管理"], prefix='/goods')
  user_router = APIRouter(tags=["用户管理"])@goods_router.get('/info')
  async def info():return '商品内容获取成功'@user_router.get('/info')
  async def info():return '用户内容获取成功'@user_router.get('/login')
  async def login():return '用户登录成功'# 添加路由器
  app.include_router(goods_router)
  app.include_router(user_router,  prefix='/user')
  

• 多路由嵌套

  from fastapi import APIRouter
  router = APIRouter() # 创建路由器v1_router = APIRouter(prefix='/v1/api')
  goods_router = APIRouter(tags=["商品管理"], prefix='/goods')
  user_router = APIRouter(tags=["用户管理"], prefix='/user')@goods_router.get('/info')
  async def info():return '商品内容获取成功'@user_router.get('/info')
  async def info():return '用户内容获取成功'@user_router.get('/login')
  async def login():return '用户登录成功'# 添加路由器v1_router.include_router(goods_router)
  v1_router.include_router(user_router)
  app.include_router(v1_router)
  

  • 效果：

    

• 问答：

  1. APIRouter的主要作用是什么?

     A、替代FastAPI主应用实例

     B、实现数据库连接池

     C、模块化组织路由端点

     D、生成前端用户界面

  2. 如何为路由器下所有路由添加 /api/v1 前缀,比较好?

     A、在每个路由装饰器手动添加前缀

     B、使用 APIRouter(prefix=“/api/v1”)

     C、修改FastAPI应用的mount路径

     D、在中间件中重写请求路径

5.4. 项目结构

企业级项目通常需要遵循标准化的代码组织规范，拆分文件是行业最佳实践之一。这种结构也便于集成自动化测试、CI/CD 流程 和代码审查。

• 软件功能目录结构示例

  project/
  ├── main.py	# 主入口文件，初始化 FastAPI 应用并集成路由、中间件
  ├── config/
  │   └── database.py	# 数据库配置和Tortoise-ORM 初始化
  ├── models/
  │   └── user.py	# 数据模型定义（如User 模型）
  ├── schemas/
  │   └── user.py	# Pydantic 模式定义，用于输入输出验证
  ├── routers/
  │   ├── __init__.py	# 标记 routers为Python 包
  │   ├── user.py		# 用户相关的API路由
  │   └── item.py     # 商品相关的API路由
  └── middleware/└── user_middleware.py		# 自定义中间件逻辑
  

• 按业务模块的目录结构示例

  project/
  ├── main.py		# 主入口文件，初始化 FastAPI 应用
  ├── config/
  │   └── database.py		# 数据库配置和Tortoise-ORM初始化
  ├── modules/ 	# 业务模块目录               
  │   ├── user/		# 用户管理模块
  │   │   ├── __init__.py    # 标记 user 为 Python 包
  │   │   ├── models.py      # 用户相关的数据模型
  │   │   ├── schemas.py     # 用户相关的Pydantic模式
  │   │   └── routers.py     # 用户相关的API路由
  │   ├── item/	# 商品管理模块
  │   │   ├── __init__.py    # 标记 item为Python 包
  │   │   ├── models.py      # 商品相关的数据模型 （当前为空）
  │   │   ├── schemas.py     # 商品相关的Pydantic 模式（当前为空）
  │   │   └── routers.py     # 商品相关的API路由
  └── middleware/└── user_middleware.py    # 自定义中间件逻辑
  

5.5. 项目发布

• 打包

  1. 打包为exe单文件

     • 安装PyInstaller

       pip install pyinstaller
       

       打包为一个文件

       pyinstaller --onefile main.py 
       

       如果reload的值为reload=False（默认为）

       uvicorn.run('main:app', host='127.0.0.1', port=8000, reload=False)
       

       则打包命令

       pyinstaller --onefile --add-data "main.py;." main.py
       

       • 效果图：

         

  2. 其他打包方式我遇到再写

---

1. A-Z ↩︎