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

学习!FastAPI

目录

  • FastAPI简介
  • 快速开始
    • 安装FastApi
    • FastAPI CLI
    • 自动化文档
  • Reqeust
    • 路径参数
      • Enum 类用于路径参数
      • 路径参数和数值校验
    • 查询参数
      • 查询参数和字符串校验
    • 请求体
      • 多个请求体参数
      • 嵌入单个请求体参数
    • Cookie
    • Header
    • 表单
    • 文件
    • 直接使用请求
  • Response
    • Response Model
      • 多个关联模型
    • 响应状态码
    • 处理错误
      • 自定义异常处理
    • 直接返回响应
      • HTML,流,文件和其他
    • Cookies
    • 响应头
  • 路径操作
    • 路径操作配置
    • 依赖项
      • 类作为依赖项
      • 子依赖项
      • 路径操作装饰器依赖项
      • 全局依赖项
      • 使用yield的依赖项
    • 高级依赖项
    • 子路由APIRouter
    • 为什么这样配置?
  • 中间件
    • CORS(跨域资源共享)
    • 其他中间件
    • 和依赖项的区别
  • 后台任务
  • 生命周期事件
  • SQL(关系型)数据库——现代化ORM
  • 静态文件
    • 模板
  • 部署

FastAPI简介

FastAPI 是一个现代、快速(高性能)的 Web 框架,用于构建基于 Python 的 API。它是一个开源项目,基于 Starlette 和 Pydantic 库构建而成,提供了强大的功能和高效的性能 。其核心特征如下:

  • 高性能:FastAPI 基于异步编程模型(使用 Python 的 asyncawait 关键字),利用了 Python 的异步生态系统,提供出色的性能和吞吐量 。
  • 自动文档生成:FastAPI 可以自动为你的 API 生成交互式文档,支持自动检测请求参数和响应模型,并生成相应的 API 文档 。
  • 数据验证和转换:FastAPI 使用 Pydantic 库,提供了强大的数据验证和转换功能,能够自动处理请求和响应数据的验证、转换和序列化 。
  • 类型提示:FastAPI 基于 Python 的类型提示机制,提供了强类型的请求和响应处理,这样可以减少很多常见的错误,并提供更好的代码提示和可读性 。
  • 安全认证:FastAPI 支持常用的认证方式,如 OAuth2、JWT 等,并提供了对 HTTPS 的支持,可以保护你的 API 通信安全 。
  • 强大的生态系统:FastAPI 可以与众多 Python 生态系统中的工具和库无缝集成,如 SQLAlchemy、Databases、Redis 等 。

与 Django 和 Flask 的对比

特性DjangoFlaskFastAPI
框架类型全功能框架,包含 ORM、模板引擎等微框架,灵活轻量微框架,专注于构建高性能 API
异步支持原生不支持异步,需要额外配置原生不支持异步,需要额外配置原生支持异步编程
类型提示支持支持有限,主要依赖第三方库支持有限,主要依赖第三方库原生支持,利用 Python 的类型提示
自动文档生成需要手动配置或使用第三方库需要手动配置或使用第三方库自动生成,支持 Swagger UI 和 ReDoc
性能较低,适合中小型项目中等,适合小型项目高性能,适合构建高并发、高性能的 API 服务
学习曲线陡峭,需要学习完整的框架结构平缓,适合初学者平缓,适合有一定 Python 基础的开发者
社区支持成熟,拥有庞大的社区和丰富的资源成熟,拥有活跃的社区和丰富的插件新兴,社区在快速增长,资源逐渐丰富

为什么学习 FastAPI 是必要的?

  1. 现代 Web 开发的需求:随着前后端分离架构的流行,构建高性能的 RESTful API 成为主流需求。FastAPI 专注于 API 的构建,满足现代 Web 开发的需求。

  2. 高性能的优势:FastAPI 的异步支持和高性能特性,使其在处理高并发请求时表现出色,适合构建需要高吞吐量的应用,如实时数据处理、机器学习模型部署等。

  3. 提升开发效率:自动文档生成、类型提示支持和数据验证功能,减少了重复性工作,提高了开发效率和代码质量。

  4. 良好的可维护性:强类型支持和清晰的代码结构,使得项目更易于维护和扩展,适合长期发展的项目。

  5. 快速增长的社区:虽然 FastAPI 是一个相对较新的框架,但其社区在快速增长,越来越多的企业和开发者开始采用 FastAPI,学习它有助于跟上技术发展的步伐。

最后在部署 Python Web 应用时,选择合适的服务器对性能和扩展性至关重要。以下是对 Flask 和 FastAPI 在部署方面的对比,特别关注 Gunicorn、uWSGI 和 Uvicorn 的性能表现。

  • Gunicorn
  • 优点:成熟稳定,配置简单,适合中等负载的应用。
  • 缺点:在高并发场景下,性能可能下降,延迟增加。
  • uWSGI
  • 优点:功能丰富,支持多线程,适合处理大量请求。
  • 缺点:配置复杂,调优难度较大。

实测数据显示,Flask 在使用 Gunicorn 或 uWSGI 部署时的吞吐量和延迟表现不如异步框架。

FastAPI 是基于 ASGI 的现代异步框架,常与 Uvicorn 或 Gunicorn(结合 UvicornWorker)部署,具备更高的性能和并发处理能力。(scien.cx)

  • Uvicorn
  • 优点:轻量高效,原生支持异步,适合小型或中等规模的应用。
  • 缺点:在大型生产环境中,可能缺乏进程管理等高级功能。(scien.cx)
  • Gunicorn + UvicornWorker
  • 优点:结合了 Gunicorn 的进程管理和 Uvicorn 的异步性能,适合高并发、大规模的生产环境。
  • 缺点:配置相对复杂,需要额外安装 UvicornWorker。(scien.cx, GitHub)

实测数据显示,FastAPI 在使用 Gunicorn + UvicornWorker 部署时,吞吐量和响应时间优于 Flask 的部署方式。

框架部署方式吞吐量(req/s)平均响应时间(ms)适用场景
FlaskGunicorn约 3,500较高中小型应用,低并发
FlaskuWSGI略高于 Gunicorn略低于 Gunicorn需要多线程支持的应用
FastAPIUvicorn约 11,000较低中等规模,异步应用
FastAPIGunicorn + UvicornWorker约 20,000更低高并发,大型生产环境(kisspeter.github.io, scien.cx, Gist, GitHub, php.cn, Medium, GitHub)

通过以上数据对比,我们可以直观的看出FastApi性能之强悍,尤其是现在AI时代,得益于python在机器学习方面的主场优势,很多AI平台应用都是基于langChain+FastApi+AI+SSE 去部署,AI的迅猛发展无疑为FastApi注入了新鲜血液。

其次在AI时代,Web开发已经不再局限于以往的条条框框,我们更偏向于将社区的高性能组件自行拼接达到1 x 1 x 1 … x1 >= 1的效果。而非一家独大,如django框架,当初我学习django的时候最新版本号是3.x,现在已经达到了5.x,django逐渐引入了异步视图,优化了orm和forms组件… 即便如此forms不如pydantic,orm在现代非结构化数据库中没有一丝用武之地。厚重的框架约束了开发者的思想和创造力, 随即我将目光投向了Flask!

Flask是一个微框架,在python协程还处在feature的岁月,Flask热火朝天,他的高自由度很快吸引一些中高等水平开发者的眼球,uwsgi/gunicorn + Flask部署的服务很快占到主流(Flask是wsgi框架,如果想要异步视图,需要先用 asgiref中转,再用uvicron部署)。 好景不长,python asyncio迅猛发展,anyio,asyncior,uvicorn…等高性能组件如雨后春笋一般现世,终于来到了FastApi的世界!

看到这里,也就剩下了一个念头,抛弃历史包袱,学习FastAPI!

快速开始

FastApi官方提供了非常全面的中文文档:https://fastapi.tiangolo.com/zh/learn/

Starlette (和 FastAPI) 是基于 AnyIO 实现的,这使得它们可以兼容 Python 的标准库 asyncio 和 Trio。(通过 FastAPI 你可以获得所有 Starlette 的特性 ( FastAPI 就像加强版的 Starlette ) )

特别是,你可以直接使用 AnyIO 来处理高级的并发用例,这些用例需要在自己的代码中使用更高级的模式。

即使你没有使用 FastAPI,你也可以使用 AnyIO 编写自己的异步程序,使其拥有较高的兼容性并获得一些好处(例如, 结构化并发)。

我(指原作者)基于 AnyIO 新建了一个库,作为一个轻量级的封装层,用来优化类型注解,同时提供了更好的自动补全、内联错误提示等功能。这个库还附带了一个友好的入门指南和教程,能帮助你理解并编写自己的异步代码:Asyncer。如果你有结合使用异步代码和常规(阻塞/同步)代码的需求,这个库会特别有用。

安装FastApi

那么现在你就可以通过pip去安装FastAPI:pip install "fastapi[standard]"

有人这时候比较疑惑,standard是做什么的?

在 Python 的包管理中,方括号 [] 表示安装额外的可选依赖项。因此,fastapi[standard] 会安装 FastAPI 的核心功能以及被标记为 standard 的可选依赖项。

因此pip install "fastapi[standard]"pip install fastapi 之间存在显著区别。前者安装了 FastAPI 的核心功能以及一组常用的可选依赖项,而后者仅安装了 FastAPI 的核心功能。(fastapi.org.cn)

根据 FastAPI 的官方文档,standard 组包含以下额外依赖项:

  • email-validator:用于验证电子邮件地址的格式。
  • httpx:用于测试客户端(TestClient)的 HTTP 请求。
  • jinja2:用于模板渲染。
  • python-multipart:用于处理表单数据(request.form())。
  • uvicorn:ASGI 服务器,用于运行 FastAPI 应用。
  • fastapi-cli:提供命令行工具 fastapi。 (pypi.ac.cn, fastapi.org.cn)

这些依赖项使得 FastAPI 更加功能完整,适用于开发和测试阶段。(fastapi.org.cn)

  • 仅安装核心功能:如果您只需要 FastAPI 的基本功能,可以使用:

    pip install fastapi
    
  • 安装标准依赖项:如果您需要上述提到的额外功能,可以使用:

    pip install "fastapi[standard]"
    

FastAPI CLI

FastAPI CLI 是一个命令行程序,你可以用它来部署和运行你的 FastAPI 应用程序,管理你的 FastAPI 项目,等等。

要在开发环境中运行你的 FastAPI 应用,你可以使用 fastapi dev 命令:

fastapi dev main.py

该命令行程序 fastapi 就是 FastAPI CLI。FastAPI CLI 接收你的 Python 程序路径,自动检测包含 FastAPI 的变量(通常命名为 app)及其导入方式,然后启动服务。

在生产环境中,你应该使用 fastapi run 命令。🚀

在内部,**FastAPI CLI 使用了 Uvicorn,这是一个高性能、适用于生产环境的 ASGI 服务器。**😎

注意:当你运行 fastapi dev 时,它将以开发模式运行。默认情况下,它会启用自动重载,因此当你更改代码时,它会自动重新加载服务器。该功能是资源密集型的,且相较不启用时更不稳定,因此你应该仅在开发环境下使用它。

废话不多说,我们开始第一行FastApi代码:

from fastapi import FastAPIapp = FastAPI()@app.get("/")
async def root():return {"message": "Hello World"}

运行实时服务器:fastapi dev main.py

FastAPI   Starting development server 🚀Searching for package file structure from directorieswith __init__.py filesImporting from /home/user/code/awesomeappmodule   🐍 main.pycode   Importing the FastAPI app object from the module withthe following code:from main import appapp   Using import string: main:appserver   Server started at http://127.0.0.1:8000server   Documentation at http://127.0.0.1:8000/docstip   Running in development mode, for production use:fastapi runLogs:INFO   Will watch for changes in these directories:['/home/user/code/awesomeapp']INFO   Uvicorn running on http://127.0.0.1:8000 (Press CTRL+Cto quit)INFO   Started reloader process [383138] using WatchFilesINFO   Started server process [383153]INFO   Waiting for application startup.INFO   Application startup complete.

打开浏览器访问 http://127.0.0.1:8000,你将看到如下的 JSON 响应:

{"message": "Hello World"}

自动化文档

跳转到 http://127.0.0.1:8000/docs,你将会看到自动生成的交互式 API 文档(由 Swagger UI 提供):

在这里插入图片描述
前往 http://127.0.0.1:8000/redoc,你将会看到可选的自动生成文档 (由 ReDoc 提供):

在这里插入图片描述

在 FastAPI 中,术语“交互式 API 文档”和“可选的 API 文档”通常指的是两种不同风格的自动生成的文档界面:Swagger UIReDoc。这两者都基于 OpenAPI 规范,提供了交互式的 API 文档体验,但在界面设计和功能侧重点上有所不同。(DeepWiki, 菜鸟教程)

🔍 交互式 API 文档(Swagger UI)

  • 访问地址:默认情况下,运行 FastAPI 应用后,可以通过 http://127.0.0.1:8000/docs 访问。(菜鸟教程)

  • 主要特点

    • 交互性强:提供“Try it out”按钮,允许用户直接在浏览器中测试 API 端点。
    • 实时反馈:在文档中填写参数并执行请求后,立即显示响应结果,方便调试。
    • 界面直观:以卡片形式展示每个端点,结构清晰,便于浏览。(菜鸟教程, 博客园)
  • 适用场景:开发阶段频繁测试 API、需要快速验证请求和响应的场景。(Traffine I/O)

  • 📘 可选的 API 文档(ReDoc)
  • 访问地址:默认情况下,运行 FastAPI 应用后,可以通过 http://127.0.0.1:8000/redoc 访问。(菜鸟教程)

  • 主要特点

    • 文档导向:侧重于提供详细的 API 描述,适合生成正式的 API 文档。
    • 界面简洁:采用单栏布局,目录结构清晰,便于查阅。
    • 缺乏交互测试:不支持直接在文档中测试 API 请求。(菜鸟教程)
  • 适用场景:需要为用户或第三方开发者提供正式、可读性强的 API 文档的场景。

特性Swagger UI(/docs)ReDoc(/redoc)
交互性✅ 支持直接测试 API 请求❌ 不支持交互测试
文档结构📄 卡片式展示,适合快速浏览和测试📘 侧边目录,适合详细查阅
界面风格🎨 现代化、功能导向🧾 简洁、文档导向
适用阶段🧪 开发和调试阶段📚 正式文档发布阶段(FastAPI)

✅ 建议

  • 开发阶段:使用 Swagger UI(/docs)进行 API 测试和调试。
  • 文档发布:使用 ReDoc(/redoc)生成正式的 API 文档供用户查阅。(菜鸟教程)

FastAPI 通过这两种文档界面,满足了开发者在不同阶段的需求,提升了开发效率和文档质量。

Reqeust

在 FastAPI 中,Request 对象是处理 HTTP 请求的核心组件。它并非 FastAPI 原生实现,而是直接继承自 Starlette 框架的 Request 类。Starlette 是一个轻量级的 ASGI(Asynchronous Server Gateway Interface)框架,专为构建异步 Web 应用设计。FastAPI 基于 Starlette 构建,提供了数据验证、依赖注入和自动生成 API 文档等高级功能。 (知乎专栏, biaodianfu.com)

在 Starlette 中,Request 对象的初始化方式如下:

class Request:def __init__(self, scope, receive):self.scope = scopeself._receive = receiveself._stream_consumed = False

这里的 scope 是一个包含请求信息的字典,符合 ASGI 规范,包含了请求的方法、路径、头信息等。receive 是一个异步函数,用于接收请求体的数据。这种设计使得 Request 对象能够在异步环境中高效地处理 HTTP 请求。

与 Flask 不同,FastAPI(通过 Starlette)并未使用 threading.localcontextvars 来管理请求上下文。相反,它依赖于显式传递的 scopereceive,避免了线程或协程本地存储的复杂性,提升了性能和可维护性。

路径参数

FastAPI 支持使用 Python 字符串格式化语法声明路径参数(变量):

from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_item(item_id):return {"item_id": item_id}

这段代码把路径参数 item_id 的值传递给路径函数的参数 item_id。

还可以使用 Python 标准类型注解,声明路径操作函数中路径参数的类型(复杂参数校验应该交给pydantic!):

@app.get("/items/{item_id}")
async def read_item(item_id: int):return {"item_id": item_id}

运行示例并访问 http://127.0.0.1:8000/items/3注意url位置参数是字符串哦),返回的响应如下:

{"item_id":3}

可以看到,FastAPI 通过类型声明自动解析请求中的数据,对于不能类型转换的参数(访问:http://127.0.0.1:8000/items/foo)则会报错:

在这里插入图片描述
顺序很重要,有时,路径操作中的路径是写死的。

  1. 比如要使用 /users/me 获取当前用户的数据。

  2. 然后还要使用 /users/{user_id},通过用户 ID 获取指定用户的数据。

由于路径操作是按顺序依次运行的,因此,一定要在 /users/{user_id} 之前声明 /users/me

from fastapi import FastAPIapp = FastAPI()@app.get("/users/me")
async def read_user_me():return {"user_id": "the current user"}@app.get("/users/{user_id}")
async def read_user(user_id: str):return {"user_id": user_id}

否则,/users/{user_id} 将匹配 /users/me,FastAPI 会认为正在接收值为 “me” 的 user_id 参数。

Enum 类用于路径参数

路径操作使用 Python 的 Enum 类型接收预设的路径参数。

导入 Enum 并创建继承自 str 和 Enum 的子类。通过从 str 继承,API 文档就能把值的类型定义为字符串,并且能正确渲染。

然后,创建包含固定值的类属性,这些固定值是可用的有效值:

from enum import Enumfrom fastapi import FastAPIclass ModelName(str, Enum):alexnet = "alexnet"resnet = "resnet"lenet = "lenet"app = FastAPI()@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):if model_name is ModelName.alexnet:return {"model_name": model_name, "message": "Deep Learning FTW!"}if model_name.value == "lenet":return {"model_name": model_name, "message": "LeCNN all the images"}return {"model_name": model_name, "message": "Have some residuals"}

API 文档会显示预定义路径参数的可用值:

在这里插入图片描述
可以使用 model_name.value 或 your_enum_member.value 或 ModelName.lenet.value 获取实际的值(本例中为字符串)!

路径参数和数值校验

你可以使用 Path 为路径参数声明型校验和元数据。

from typing import Annotatedfrom fastapi import FastAPI, Path, Queryapp = FastAPI()@app.get("/items/{item_id}")
async def read_items(item_id: Annotated[int, Path(title="The ID of the item to get")],q: Annotated[str | None, Query(alias="item-query")] = None,
):results = {"item_id": item_id}if q:results.update({"q": q})return results

对 FastAPI 来说参数顺序无关紧要。它将通过参数的名称、类型和默认值声明(Query、Path 等)来检测参数,而不在乎参数的顺序。

因此,你可以将函数声明为:

from fastapi import FastAPI, Pathapp = FastAPI()@app.get("/items/{item_id}")
async def read_items(q: str, item_id: int = Path(title="The ID of the item to get")):results = {"item_id": item_id}if q:results.update({"q": q})return results

查询参数

声明的参数不是路径参数时,路径操作函数会把该参数自动解释为查询参数。查询字符串是键值对的集合,这些键值对位于 URL 的 ? 之后,以 & 分隔。例如,以下 URL 中:

http://127.0.0.1:8000/items/?skip=0&limit=10

查询参数为:

  • skip:值为 0
  • limit:值为 10

这些值都是 URL 的组成部分,因此,它们的类型本应是字符串。但声明 Python 类型(上例中为 int)之后,这些值就会转换为声明的类型,并进行类型校验。

from fastapi import FastAPIapp = FastAPI()fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):return fake_items_db[skip : skip + limit]

查询参数不是路径的固定内容,它是可选的,还支持默认值。上例用 skip=0 和 limit=10 设定默认值。

访问 URL:

http://127.0.0.1:8000/items/

与访问以下地址相同:

http://127.0.0.1:8000/items/?skip=0&limit=10

同理,把默认值设为 None 即可声明可选的查询参数,这其实是一种联合类型联合类型(Union types),它的意思是变量 q 的类型可以是 str 或者 None,也就是可选的字符串。:

@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None):if q:return {"item_id": item_id, "q": q}return {"item_id": item_id}

这是 Python 3.10+ 引入的 联合类型 简写形式,相当于:

from typing import Unionq: Union[str, None] = None

python引入类型注解后,我们会经常见到C类型及其扩展类型,标注类型不得不说是一个非常好的开发习惯,尽管Python解释器并不会校验类型:

概念示例用途
联合类型`strNoneUnion[str, None]`变量可以是多个类型中的一个
泛型List[int], Dict[str, Any]参数化类型,如列表中元素是 int
类型别名UserID = int为某个类型起一个新名字,增加可读性

参数还可以声明为 bool 类型,FastAPI 会自动转换参数类型:

from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None, short: bool = False):item = {"item_id": item_id}if q:item.update({"q": q})if not short:item.update({"description": "This is an amazing item that has a long description"})return item

本例中,访问:

http://127.0.0.1:8000/items/foo?short=1
或http://127.0.0.1:8000/items/foo?short=True
或http://127.0.0.1:8000/items/foo?short=true
或http://127.0.0.1:8000/items/foo?short=on
或http://127.0.0.1:8000/items/foo?short=yes

或其它任意大小写形式(大写、首字母大写等),函数接收的 short 参数都是布尔值 True。值为 False 时也一样。

FastAPI 可以识别同时声明的多个路径参数和查询参数,而且声明查询参数的顺序并不重要。

from fastapi import FastAPIapp = FastAPI()@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: str, q: str | None = None, short: bool = False
):item = {"item_id": item_id, "owner_id": user_id}if q:item.update({"q": q})if not short:item.update({"description": "This is an amazing item that has a long description"})return item

如果要把查询参数设置为必选,就不要声明默认值:

from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str):item = {"item_id": item_id, "needy": needy}return item

查询参数和字符串校验

FastAPI 允许你为参数声明额外的信息和校验,比如我们打算添加约束条件:即使 q 是可选的,但只要提供了该参数,则该参数值不能超过50个字符的长度。

为此,首先从 fastapi 导入 Query:

from typing import Unionfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, max_length=50)):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results

from typing import Annotatedfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Annotated[str | None, Query(max_length=50)] = None):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results

Annotated 是 Python 3.9+ 引入的一个类型提示增强工具,主要作用是给类型添加“元信息”(metadata),使得类型提示不仅表达类型本身,还能携带额外说明。

你还可以添加 min_length 参数,正则表达式:

@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, min_length=3, max_length=50, pattern="^fixedquery$"),
):

当你使用 Query 显式地定义查询参数时,你还可以声明它去接收一组值,或换句话来说,接收多个值。

@app.get("/items/")
async def read_items(q: Union[List[str], None] = Query(default=None)):query_items = {"q": q}return query_items

然后请求http://localhost:8000/items/?q=foo&q=bar,你会在路径操作函数的函数参数 q 中以一个 Python list 的形式接收到查询参数 q 的多个值。

请求体

使用 Pydantic 模型声明请求体,能充分利用它的功能和优点。

from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneapp = FastAPI()@app.post("/items/")
async def create_item(item: Item):return item

与声明查询参数一样,包含默认值的模型属性是可选的,否则就是必选的。默认值为 None 的模型属性也是可选的。

仅使用 Python 类型声明,FastAPI 就可以:

  • 以 JSON 形式读取请求体:(在必要时)把请求体转换为对应的类型
  • 校验数据:数据无效时返回错误信息,并指出错误数据的确切位置和内容
  • 把接收的数据赋值给参数 item,把函数中请求体参数的类型声明为 Item,还能获得代码补全等编辑器支持
  • 为模型生成 JSON Schema,在项目中所需的位置使用。

FastAPI 支持同时声明路径参数和请求体,FastAPI 能识别与路径参数匹配的函数参数,还能识别从请求体中获取的类型为 Pydantic 模型的函数参数。

from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneapp = FastAPI()@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):return {"item_id": item_id, **item.dict()}

FastAPI 还支持同时声明请求体、路径参数和查询参数。

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: str | None = None):result = {"item_id": item_id, **item.dict()}if q:result.update({"q": q})return result

函数参数按如下规则进行识别:

  • 路径中声明了相同参数的参数,是路径参数
  • 类型是(int、float、str、bool 等)单类型的参数,是查询参数
  • 类型是 Pydantic 模型的参数,是请求体

多个请求体参数

毫无疑问地,你可以随意地混合使用 Path、Query 和请求体参数声明,FastAPI 会知道该如何处理。

from typing import Annotatedfrom fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneclass User(BaseModel):username: strfull_name: str | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):results = {"item_id": item_id, "item": item, "user": user, "importance": importance}return results

在这种情况下,FastAPI 将期望像这样的请求体:

{"item": {"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2},"user": {"username": "dave","full_name": "Dave Grohl"},"importance": 5
}

与使用 Query 和 Path 为查询参数和路径参数定义额外数据的方式相同,FastAPI 提供了一个同等的 Body,例如,为了扩展先前的模型,你可能决定除了 item 和 user 之外,还想在同一请求体中具有另一个键 importance。

如果你就按原样声明它,因为它是一个单一值,FastAPI 将假定它是一个查询参数。但是你可以使用 Body 指示 FastAPI 将其作为请求体的另一个键进行处理。

嵌入单个请求体参数

假设你只有一个来自 Pydantic 模型 Item 的请求体参数 item。

默认情况下,FastAPI 将直接期望这样的请求体。

但是,如果你希望它期望一个拥有 item 键并在值中包含模型内容的 JSON,就像在声明额外的请求体参数时所做的那样,则可以使用一个特殊的 Body 参数 embed:

item: Item = Body(embed=True)

比如:

from typing import Annotatedfrom fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):results = {"item_id": item_id, "item": item}return results

在这种情况下,FastAPI 将期望像这样的请求体:

{"item": {"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2}
}

而不是:

{"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2
}

Cookie

定义 Cookie 参数与定义 Query 和 Path 参数一样。

from typing import Annotatedfrom fastapi import Cookie, FastAPIapp = FastAPI()@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):return {"ads_id": ads_id}

如果您有一组相关的 cookie,您可以创建一个 Pydantic 模型来声明它们。

from typing import Annotatedfrom fastapi import Cookie, FastAPI
from pydantic import BaseModelapp = FastAPI()class Cookies(BaseModel):session_id: strfatebook_tracker: str | None = Nonegoogall_tracker: str | None = None@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):return cookies

FastAPI 将从请求中接收到的 cookie 中提取出每个字段的数据,并提供您定义的 Pydantic 模型。

注意:你不能直接写 cookies: Cookies,因为 FastAPI 必须通过 Cookie() 明确告诉它从 cookie 中读取这些字段,否则它不知道你想从哪里获取这个 Cookies 模型的字段。

这是 FastAPI 的一个核心设计原则:参数类型 + 额外标记(如 Query, Header, Cookie)共同决定参数来源。

在某些特殊使用情况下(可能并不常见),您可能希望限制您想要接收的 cookie。您可以使用 Pydantic 的模型配置来禁止( forbid )任何额外( extra )字段:

from typing import Annotated, Unionfrom fastapi import Cookie, FastAPI
from pydantic import BaseModelapp = FastAPI()class Cookies(BaseModel):model_config = {"extra": "forbid"}session_id: strfatebook_tracker: Union[str, None] = Nonegoogall_tracker: Union[str, None] = None@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):return cookies

如果客户尝试发送一些额外的 cookie,他们将收到错误响应。例如,如果客户端尝试发送一个值为 good-list-please 的 santa_tracker cookie,客户端将收到一个错误响应,告知他们 santa_tracker cookie 是不允许的:

{"detail": [{"type": "extra_forbidden","loc": ["cookie", "santa_tracker"],"msg": "Extra inputs are not permitted","input": "good-list-please",}]
}

Header

定义 Header 参数的方式与定义 Query、Path、Cookie 参数相同。

from typing import Annotatedfrom fastapi import FastAPI, Headerapp = FastAPI()@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):return {"User-Agent": user_agent}

Header 比 Path、Query 和 Cookie 提供了更多功能。大部分标准请求头用连字符分隔,即减号(-)。

但是 user-agent 这样的变量在 Python 中是无效的。因此,默认情况下,Header 把参数名中的字符由下划线(_)改为连字符(-)来提取并存档请求头 。

同时,HTTP 的请求头不区分大小写,可以使用 Python 标准样式(即 snake_case)进行声明。因此,可以像在 Python 代码中一样使用 user_agent ,无需把首字母大写为 User_Agent 等形式。

如需禁用下划线自动转换为连字符,可以把 Header 的 convert_underscores 参数设置为 False:

from typing import Annotatedfrom fastapi import FastAPI, Headerapp = FastAPI()@app.get("/items/")
async def read_items(strange_header: Annotated[str | None, Header(convert_underscores=False)] = None,
):return {"strange_header": strange_header}

如果您有一组相关的 header 参数,您可以创建一个 Pydantic 模型来声明它们。

from typing import Annotatedfrom fastapi import FastAPI, Header
from pydantic import BaseModelapp = FastAPI()class CommonHeaders(BaseModel):host: strsave_data: boolif_modified_since: str | None = Nonetraceparent: str | None = Nonex_tag: list[str] = []@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):return headers

表单

接收的不是 JSON,而是表单字段时,要使用 Form(需要提前下载pip install python-multipart)。

from fastapi import FastAPI, Formapp = FastAPI()@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):return {"username": username}

创建表单(Form)参数的方式与 Body 和 Query 一样,例如,OAuth2 规范的 “密码流” 模式规定要通过表单字段发送 username 和 password。该规范要求字段必须命名为 username 和 password,并通过表单字段发送,不能用 JSON。

使用 Form 可以声明与 Body (及 Query、Path、Cookie)相同的元数据和验证。

您可以使用 Pydantic 模型在 FastAPI 中声明表单字段。您只需声明一个 Pydantic 模型,其中包含您希望接收的表单字段,然后将参数声明为 Form :

from typing import Annotatedfrom fastapi import FastAPI, Form
from pydantic import BaseModelapp = FastAPI()class FormData(BaseModel):username: strpassword: str@app.post("/login/")
async def login(data: Annotated[FormData, Form()]):return data

文件

File 用于定义客户端的上传文件,因为上传文件以「表单数据」形式发送,所以接收上传文件,要预先安装 python-multipart。

from fastapi import FastAPI, File, UploadFileapp = FastAPI()@app.post("/files/")
async def create_file(file: bytes = File()):return {"file_size": len(file)}@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):return {"filename": file.filename}

如果把路径操作函数参数的类型声明为 bytes,FastAPI 将以 bytes 形式读取和接收文件内容。这种方式把文件的所有内容都存储在内存里,适用于小型文件。不过,很多情况下,UploadFile 更好用。

UploadFile 与 bytes 相比有更多优势:

  • 使用 spooled 文件:存储在内存的文件超出最大上限时,FastAPI 会把文件存入磁盘;
  • 这种方式更适于处理图像、视频、二进制文件等大型文件,好处是不会占用所有内存;
  • 可获取上传文件的元数据;
  • 自带 file-like async 接口;
  • 暴露的 Python SpooledTemporaryFile 对象,可直接传递给其他预期「file-like」对象的库。

UploadFile 对象包含以下常用属性:

属性名类型描述
filenamestr上传文件的文件名,例如 "myimage.jpg"
content_typestr文件的 MIME 类型,例如 "image/jpeg"
fileSpooledTemporaryFile(类文件对象)可像普通 Python 文件一样使用,支持传递给支持 file-like 的其他库

UploadFile 提供了一系列基于异步的文件操作方法,这些方法底层由 SpooledTemporaryFile 提供支持:

方法名参数描述
await write(data)strbytes将数据写入文件
await read(size)int(可选)读取指定大小(字节数)的内容,默认读取全部
await seek(offset)int移动文件指针至指定位置,例如 await file.seek(0) 移动到文件开头
await close()关闭文件对象

异步读取示例(在 async def 中)

@app.post("/upload/")
async def upload_file(myfile: UploadFile):contents = await myfile.read()  # 异步读取上传内容return {"filename": myfile.filename, "content": contents.decode()}

同步读取示例(在 def 中)

@app.post("/upload/")
def upload_file(myfile: UploadFile):contents = myfile.file.read()  # 同步读取return {"filename": myfile.filename, "content": contents.decode()}

💡 注意事项

  • UploadFile 是 FastAPI 推荐用于处理大文件上传的方式,优于 bytes,因为它不会将整个文件内容加载到内存中。
  • 方法为异步(async),必须在 async def 函数中使用 await
  • 如果你用的是同步路径操作函数(def),可以直接访问 .file,像普通文件对象一样操作。
  • 使用 async 方法时,FastAPI 在线程池中执行文件方法,并 await 操作完成。

您可以通过使用标准类型注解并将 None 作为默认值的方式将一个文件参数设为可选:

from fastapi import FastAPI, File, UploadFileapp = FastAPI()@app.post("/files/")
async def create_file(file: bytes | None = File(default=None)):if not file:return {"message": "No file sent"}else:return {"file_size": len(file)}@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile | None = None):if not file:return {"message": "No upload file sent"}else:return {"filename": file.filename}

您也可以将 File() 与 UploadFile 一起使用,例如,设置额外的元数据:

from fastapi import FastAPI, File, UploadFileapp = FastAPI()@app.post("/files/")
async def create_file(file: bytes = File(description="A file read as bytes")):return {"file_size": len(file)}@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(description="A file read as UploadFile"),
):return {"filename": file.filename}

FastAPI 支持同时上传多个文件。可用同一个「表单字段」发送含多个文件的「表单数据」。上传多个文件时,要声明含 bytes 或 UploadFile 的列表(List):

@app.post("/files/")
async def create_files(files: list[bytes] = File()):return {"file_sizes": [len(file) for file in files]}@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile]):return {"filenames": [file.filename for file in files]}@app.post("/files/")
async def create_files(files: list[bytes] = File(description="Multiple files as bytes"),
):return {"file_sizes": [len(file) for file in files]}@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile] = File(description="Multiple files as UploadFile"),
):return {"filenames": [file.filename for file in files]}

直接使用请求

至此,我们已经使用多种类型声明了请求的各种组件。

并从以下对象中提取数据:

  • 路径参数
  • 请求头
  • Cookies

FastAPI 使用这种方式验证数据、转换数据,并自动生成 API 文档。但有时,我们也需要直接访问 Request 对象。

FastAPI 的底层是 Starlette,FastAPI 只不过是在 Starlette 顶层提供了一些工具,所以能直接使用 Starlette 的 Request 对象。

假设要在路径操作函数中获取客户端 IP 地址和主机,此时,需要直接访问请求。

from fastapi import FastAPI, Requestapp = FastAPI()@app.get("/items/{item_id}")
def read_root(item_id: str, request: Request):client_host = request.client.hostreturn {"client_host": client_host, "item_id": item_id}

把路径操作函数的参数类型声明为 Request,FastAPI 就能把 Request 传递到参数里。

Response

在 FastAPI 中,响应(Response) 是与请求(Request)相对的核心部分,用于将数据、安全信息、状态码和内容格式传递给客户端。

FastAPI 默认会根据返回值自动生成响应内容,例如 dict 会被自动转换为 JSON 并附带合适的 Content-Type。但当你需要自定义响应格式、状态码、头部或流式响应时,可以使用 FastAPI 提供的 Response 类及其衍生类,如:

  • JSONResponse:返回 JSON 数据(默认类型);
  • HTMLResponse:返回 HTML 页面内容;
  • PlainTextResponse:返回纯文本;
  • StreamingResponse:流式传输大文件或数据;
  • RedirectResponse:用于重定向;
  • FileResponse:用于文件下载;
  • ORJSONResponse / UJSONResponse:使用更快的 JSON 序列化库以提升性能。

此外,FastAPI 允许你通过:

  • 设置响应模型(response_model):对输出内容结构进行约束与验证;
  • 使用 Response 参数:动态修改响应头、状态码、cookie 等;
  • 自定义状态码与媒体类型:精细控制返回行为。

Response Model

你可以在任意的路径操作中使用 response_model 参数来声明用于响应的模型:

  • @app.get()
  • @app.post()
  • @app.put()
  • @app.delete()
  • 等等。

并且,response_model是「装饰器」方法(get,post 等)的一个参数。不像之前的所有参数和请求体,它不属于路径操作函数。

它接收的类型与你将为 Pydantic 模型属性所声明的类型相同,因此它可以是一个 Pydantic 模型,但也可以是一个由 Pydantic 模型组成的 list,例如 List[Item]。

FastAPI 将使用此 response_model 来:

  • 将输出数据转换为其声明的类型。
  • 校验数据。
  • 在 OpenAPI 的路径操作中为响应添加一个 JSON Schema。
  • 并在自动生成文档系统中使用。
  • 将输出数据限制在该模型定义内
from typing import Anyfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonetags: list[str] = []@app.post("/items/", response_model=Item)
async def create_item(item: Item) -> Any:return item@app.get("/items/", response_model=list[Item])
async def read_items() -> Any:return [{"name": "Portal Gun", "price": 42.0},{"name": "Plumbus", "price": 32.0},]

你的响应模型可以具有默认值,例如:

from typing import List, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: float = 10.5tags: List[str] = []items = {"foo": {"name": "Foo", "price": 50.2},"bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},"baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):return items[item_id]

但如果它们并没有存储实际的值,你可能想从结果中忽略它们的默认值。你可以设置路径操作装饰器的 response_model_exclude_unset=True 参数:

@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)

然后响应中将不会包含那些默认值,而是仅有实际设置的值。

你还可以使用路径操作装饰器的 response_model_include 和 response_model_exclude 参数。它们接收一个由属性名称 str 组成的 set 来包含(忽略其他的)或者排除(包含其他的)这些属性。(可以但不建议)

多个关联模型

下面的代码展示了不同模型处理密码字段的方式,及使用位置的大致思路:

from fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp = FastAPI()class UserIn(BaseModel):username: strpassword: stremail: EmailStrfull_name: str | None = Noneclass UserOut(BaseModel):username: stremail: EmailStrfull_name: str | None = Noneclass UserInDB(BaseModel):username: strhashed_password: stremail: EmailStrfull_name: str | None = Nonedef fake_password_hasher(raw_password: str):return "supersecret" + raw_passworddef fake_save_user(user_in: UserIn):hashed_password = fake_password_hasher(user_in.password)user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)print("User saved! ..not really")return user_in_db@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):user_saved = fake_save_user(user_in)return user_saved

Pydantic 模型支持 .dict() 方法,能返回包含模型数据的字典。

FastAPI 的核心思想就是减少代码重复。代码重复会导致 bug、安全问题、代码失步等问题(更新了某个位置的代码,但没有同步更新其它位置的代码)。

上面的这些模型共享了大量数据,拥有重复的属性名和类型。

声明 UserBase 模型作为其它模型的基类。然后,用该类衍生出继承其属性(类型声明、验证等)的子类。所有数据转换、校验、文档等功能仍将正常运行。这样,就可以仅声明模型之间的差异部分(具有明文的 password、具有 hashed_password 以及不包括密码)。

通过这种方式,可以只声明模型之间的区别(分别包含明文密码、哈希密码,以及无密码的模型)。

from fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp = FastAPI()class UserBase(BaseModel):username: stremail: EmailStrfull_name: str | None = Noneclass UserIn(UserBase):password: strclass UserOut(UserBase):passclass UserInDB(UserBase):hashed_password: strdef fake_password_hasher(raw_password: str):return "supersecret" + raw_passworddef fake_save_user(user_in: UserIn):hashed_password = fake_password_hasher(user_in.password)user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)print("User saved! ..not really")return user_in_db@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):user_saved = fake_save_user(user_in)return user_saved

任意的 dict 都能用于声明响应,只要声明键和值的类型,无需使用 Pydantic 模型。事先不知道可用的字段 / 属性名时(Pydantic 模型必须知道字段是什么),这种方式特别有用。

from fastapi import FastAPIapp = FastAPI()@app.get("/keyword-weights/", response_model=dict[str, float])
async def read_keyword_weights():return {"foo": 2.3, "bar": 3.4}

响应状态码

与指定响应模型的方式相同,在以下任意路径操作中,可以使用 status_code 参数声明用于响应的 HTTP 状态码:

  • @app.get()
  • @app.post()
  • @app.put()
  • @app.delete()
  • 等……

status_code 是(get、post 等)装饰器方法中的参数。与之前的参数和请求体不同,不是路径操作函数的参数。

from fastapi import FastAPIapp = FastAPI()@app.post("/items/", status_code=201)
async def create_item(name: str):return {"name": name}

status_code 还能接收 IntEnum 类型,比如 Python 的 http.HTTPStatus。

from fastapi import FastAPI, statusapp = FastAPI()@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):return {"name": name}

在 HTTP 协议中,发送 3 位数的数字状态码是响应的一部分。这些状态码都具有便于识别的关联名称,但是重要的还是数字。

简言之:

  • 100 及以上的状态码用于返回信息。这类状态码很少直接使用。具有这些状态码的响应不能包含响应体。

  • 200 及以上的状态码用于表示成功。这些状态码是最常用的。

    • 200 是默认状态代码,表示一切正常。
    • 201 表示已创建,通常在数据库中创建新记录后使用。
    • 204 是一种特殊的例子,表示无内容。该响应在没有为客户端返回内容时使用,因此,该响应不能包含响应体。
  • 300 及以上的状态码用于重定向。具有这些状态码的响应不一定包含响应体,但 304 未修改 是个例外,该响应不得包含响应体。

  • 400 及以上的状态码用于表示客户端错误。这些可能是第二常用的类型。

    • 404 用于未找到响应。
    • 对于来自客户端的一般错误,可以只使用 400
  • 500 及以上的状态码用于表示服务器端错误。几乎永远不会直接使用这些状态码。应用代码或服务器出现问题时,会自动返回这些状态代码。

处理错误

某些情况下,需要向客户端返回错误提示。这里所谓的客户端包括前端浏览器、其他应用程序、物联网设备等。

需要向客户端返回错误提示的场景主要如下:

  • 客户端没有执行操作的权限
  • 客户端没有访问资源的权限
  • 客户端要访问的项目不存在
  • 等等 …

向客户端返回 HTTP 错误响应,可以使用 HTTPException。HTTPException 是额外包含了和 API 有关数据的常规 Python 异常。

因为是 Python 异常,所以不能 return,只能 raise。

如在调用函数里的工具函数时,触发了 HTTPException,FastAPI 就不再继续执行路径操作函数中的后续代码,而是立即终止请求,并把 HTTPException 的 HTTP 错误发送至客户端。

from fastapi import FastAPI, HTTPExceptionapp = FastAPI()items = {"foo": "The Foo Wrestlers"}@app.get("/items/{item_id}")
async def read_item(item_id: str):if item_id not in items:raise HTTPException(status_code=404, detail="Item not found")return {"item": items[item_id]}

如果客户端请求 http://example.com/items/bar(item_id 「bar」 不存在时),则会接收到 HTTP 状态码 - 404(「未找到」错误)及如下 JSON 响应结果:

{"detail": "Item not found"
}

触发 HTTPException 时,可以用参数 detail 传递任何能转换为 JSON 的值,不仅限于 str。还支持传递 dict、list 等数据结构。FastAPI 能自动处理这些数据,并将之转换为 JSON。

对于某些高级应用场景,还需要添加自定义响应头:

from fastapi import FastAPI, HTTPExceptionapp = FastAPI()items = {"foo": "The Foo Wrestlers"}@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):if item_id not in items:raise HTTPException(status_code=404,detail="Item not found",headers={"X-Error": "There goes my error"},)return {"item": items[item_id]}

自定义异常处理

添加自定义处理器,要使用 Starlette 的异常工具。

假设要触发的自定义异常叫作 UnicornException。且需要 FastAPI 实现全局处理该异常。此时,可以用 @app.exception_handler() 添加自定义异常控制器:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponseclass UnicornException(Exception):def __init__(self, name: str):self.name = nameapp = FastAPI()@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):return JSONResponse(status_code=418,content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},)@app.get("/unicorns/{name}")
async def read_unicorn(name: str):if name == "yolo":raise UnicornException(name=name)return {"unicorn_name": name}

请求 /unicorns/yolo 时,路径操作会触发 UnicornException。但该异常将会被 unicorn_exception_handler 处理。接收到的错误信息清晰明了,HTTP 状态码为 418,JSON 内容如下:

{"message": "Oops! yolo did something. There goes a rainbow..."}

FastAPI 自带了一些默认异常处理器。触发 HTTPException 或请求无效数据时,这些处理器返回默认的 JSON 响应结果。不过,也可以使用自定义处理器覆盖默认异常处理器。

请求中包含无效数据时,FastAPI 内部会触发 RequestValidationError。该异常也内置了默认异常处理器。覆盖默认异常处理器时需要导入 RequestValidationError,并用 @app.excption_handler(RequestValidationError) 装饰异常处理器。这样,异常处理器就可以接收 Request 与异常。

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):return PlainTextResponse(str(exc.detail), status_code=exc.status_code)@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):return PlainTextResponse(str(exc), status_code=400)@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id == 3:raise HTTPException(status_code=418, detail="Nope! I don't like 3.")return {"item_id": item_id}

RequestValidationError 是 Pydantic 的 ValidationError 的子类。

FastAPI 调用的就是 RequestValidationError 类,因此,如果在 response_model 中使用 Pydantic 模型,且数据有错误时,在日志中就会看到这个错误。

但客户端或用户看不到这个错误。反之,客户端接收到的是 HTTP 状态码为 500 的「内部服务器错误」。这是因为在响应或代码(不是在客户端的请求里)中出现的 Pydantic ValidationError 是代码的 bug。

同理,也可以覆盖 HTTPException 处理器。例如,只为错误返回纯文本响应,而不是返回 JSON 格式的内容:

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):return PlainTextResponse(str(exc.detail), status_code=exc.status_code)@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):return PlainTextResponse(str(exc), status_code=400)@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id == 3:raise HTTPException(status_code=418, detail="Nope! I don't like 3.")return {"item_id": item_id}

RequestValidationError 包含其接收到的无效数据请求的 body 。开发时,可以用这个请求体生成日志、调试错误,并返回给用户。

from fastapi import FastAPI, Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModelapp = FastAPI()@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):return JSONResponse(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),)class Item(BaseModel):title: strsize: int@app.post("/items/")
async def create_item(item: Item):return item

FastAPI 也提供了自有的 HTTPException。FastAPI 的 HTTPException 继承自 Starlette 的 HTTPException 错误类。

它们之间的唯一区别是,FastAPI 的 HTTPException 可以在响应中添加响应头。OAuth 2.0 等安全工具需要在内部调用这些响应头。

因此你可以继续像平常一样在代码中触发 FastAPI 的 HTTPException 。但注册异常处理器时,应该注册到来自 Starlette 的 HTTPException。

这样做是为了,当 Starlette 的内部代码、扩展或插件触发 Starlette HTTPException 时,处理程序能够捕获、并处理此异常。

FastAPI 还支持先对异常进行某些处理,然后再使用 FastAPI 中处理该异常的默认异常处理器。

从 fastapi.exception_handlers 中导入要复用的默认异常处理器:

from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import (http_exception_handler,request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):print(f"OMG! An HTTP error!: {repr(exc)}")return await http_exception_handler(request, exc)@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):print(f"OMG! The client sent invalid data!: {exc}")return await request_validation_exception_handler(request, exc)@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id == 3:raise HTTPException(status_code=418, detail="Nope! I don't like 3.")return {"item_id": item_id}

直接返回响应

当你创建一个 FastAPI 路径操作 时,你可以正常返回以下任意一种数据:dict,list,Pydantic 模型,数据库模型等等。

FastAPI 默认会使用 jsonable_encoder 将这些类型的返回值转换成 JSON 格式。然后,FastAPI 会在后台将这些兼容 JSON 的数据(比如字典)放到一个 JSONResponse 中,该 JSONResponse 会用来发送响应给客户端。但是你可以在你的 路径操作 中直接返回一个 JSONResponse。

直接返回响应可能会有用处,比如返回自定义的响应头和 cookies。

**事实上,你可以返回任意 Response 或者任意 Response 的子类。**当你返回一个 Response 时,FastAPI 会直接传递它。

FastAPI 不会用 Pydantic 模型做任何数据转换,不会将响应内容转换成任何类型,等等。由于 FastAPI 并未对你返回的 Response 做任何改变,你必须确保你已经准备好响应内容。

例如,如果不首先将 Pydantic 模型转换为 dict,并将所有数据类型(如 datetime、UUID 等)转换为兼容 JSON 的类型,则不能将其放入JSONResponse中。

from datetime import datetime
from typing import Unionfrom fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from pydantic import BaseModelclass Item(BaseModel):title: strtimestamp: datetimedescription: Union[str, None] = Noneapp = FastAPI()@app.put("/items/{id}")
def update_item(id: str, item: Item):json_compatible_item_data = jsonable_encoder(item)return JSONResponse(content=json_compatible_item_data)

假设你想要返回一个 XML 响应。你可以把你的 XML 内容放到一个字符串中,放到一个 Response 中,然后返回。

from fastapi import FastAPI, Responseapp = FastAPI()@app.get("/legacy/")
def get_legacy_data():data = """<?xml version="1.0"?><shampoo><Header>Apply shampoo here.</Header><Body>You'll have to use soap here.</Body></shampoo>"""return Response(content=data, media_type="application/xml")

FastAPI 默认使用 JSONResponse 返回一个响应,将你的视图函数中的返回内容放到该 JSONResponse 中。FastAPI 会自动使用默认的状态码或者使用你在 路径操作 中设置的状态码。如果你想要返回主要状态码之外的状态码:

return JSONResponse(status_code=status.HTTP_201_CREATED, content=item)

HTML,流,文件和其他

例如,如果你需要压榨性能,你可以安装并使用 orjson 并将响应设置为 ORJSONResponse。

from fastapi import FastAPI
from fastapi.responses import ORJSONResponseapp = FastAPI()@app.get("/items/", response_class=ORJSONResponse)
async def read_items():return ORJSONResponse([{"item_id": "Foo"}])

注意:ORJSONResponse 目前只在 FastAPI 中可用,而在 Starlette 中不可用。

如果你需要使用 HTMLResponse 来从 FastAPI 中直接返回一个 HTML 响应。导入 HTMLResponse,将 HTMLResponse 作为你的 路径操作 的 response_class 参数传入。

from fastapi import FastAPI
from fastapi.responses import HTMLResponseapp = FastAPI()@app.get("/items/", response_class=HTMLResponse)
async def read_items():return """<html><head><title>Some HTML in here</title></head><body><h1>Look ma! HTML!</h1></body></html>"""

或者也可以直接返回一个HTMLResponse:

from fastapi import FastAPI
from fastapi.responses import HTMLResponseapp = FastAPI()@app.get("/items/")
async def read_items():html_content = """<html><head><title>Some HTML in here</title></head><body><h1>Look ma! HTML!</h1></body></html>"""return HTMLResponse(content=html_content, status_code=200)

返回 HTTP 重定向。默认情况下使用 307 状态代码(临时重定向)。

from fastapi import FastAPI
from fastapi.responses import RedirectResponseapp = FastAPI()@app.get("/typer")
async def redirect_typer():return RedirectResponse("https://typer.tiangolo.com")

要传输流式传输响应主体,需要采用异步生成器或普通生成器/迭代器:

from fastapi import FastAPI
from fastapi.responses import StreamingResponseapp = FastAPI()async def fake_video_streamer():for i in range(10):yield b"some fake video bytes"@app.get("/")
async def main():return StreamingResponse(fake_video_streamer())

如果您有类似文件的对象(例如,由 open() 返回的对象),则可以在 StreamingResponse 中将其返回。包括许多与云存储,视频处理等交互的库。

from fastapi import FastAPI
from fastapi.responses import StreamingResponsesome_file_path = "large-video-file.mp4"
app = FastAPI()@app.get("/")
def main():def iterfile():  # (1)with open(some_file_path, mode="rb") as file_like:  # (2)yield from file_like  # (3)return StreamingResponse(iterfile(), media_type="video/mp4")

如果需要异步传输文件作为响应则可以使用FileResponse,

  • path - 要流式传输的文件的文件路径。
  • headers - 任何自定义响应头,传入字典类型。
  • media_type - 给出媒体类型的字符串。如果未设置,则文件名或路径将用于推断媒体类型。
  • filename - 如果给出,它将包含在响应的 Content-Disposition 中。
  • 文件响应将包含适当的 Content-Length,Last-Modified 和 ETag 的响应头。
from fastapi import FastAPI
from fastapi.responses import FileResponsesome_file_path = "large-video-file.mp4"
app = FastAPI()@app.get("/")
async def main():return FileResponse(some_file_path)

StreamingResponseFileResponse 都是 FastAPI 中用于返回文件或流式数据的响应类,但它们的使用场景和底层行为有所不同。

  • StreamingResponse
    用于返回一个流式响应,其内容可以是任何可迭代(包括生成器)或 async 可迭代对象(如异步生成器),适合用于返回动态生成的大文件或实时数据传输,节省内存开销。例如,当你要逐块读取大文件、实时编码视频流、返回日志输出等场景,可以使用 StreamingResponse

  • FileResponse
    专为返回静态文件设计。它会自动处理如文件大小、媒体类型、内容编码、浏览器缓存头等细节。适用于你已经有一个本地文件,并希望将其作为下载或直接返回给客户端的场景。底层使用 aiofiles 异步读取文件内容,性能优良且使用简单。

简而言之:

  • 如果你返回的是一个本地文件路径 —— 用 FileResponse
  • 如果你返回的是一个生成器或动态数据流 —— 用 StreamingResponse

Cookies

你可以在 路径函数 中定义一个类型为 Response的参数,这样你就可以在这个临时响应对象中设置cookie了。

from fastapi import FastAPI, Responseapp = FastAPI()@app.post("/cookie-and-object/")
def create_cookie(response: Response):response.set_cookie(key="fakesession", value="fake-cookie-session-value")return {"message": "Come to the dark side, we have cookies"}

如果你定义了 response_model,程序会自动根据response_model来过滤和转换你响应的对象。

响应头

你可以在你的路径操作函数中声明一个Response类型的参数(就像你可以为cookies做的那样)。然后你可以在这个临时响应对象中设置头部。

from fastapi import FastAPI, Responseapp = FastAPI()@app.get("/headers-and-object/")
def get_headers(response: Response):response.headers["X-Cat-Dog"] = "alone in the world"return {"message": "Hello World"}

然后你可以像平常一样返回任何你需要的对象(例如一个dict或者一个数据库模型)。如果你声明了一个response_model,它仍然会被用来过滤和转换你返回的对象。

你也可以在直接返回Response时添加头部。

from fastapi import FastAPI
from fastapi.responses import JSONResponseapp = FastAPI()@app.get("/headers/")
def get_headers():content = {"message": "Hello World"}headers = {"X-Cat-Dog": "alone in the world", "Content-Language": "en-US"}return JSONResponse(content=content, headers=headers)

路径操作

在 FastAPI 中,所谓“路径操作”(Path Operation),指的是将某个 URL 路径与一个 HTTP 方法(如 GET、POST、PUT 等)绑定,并关联到一个 Python 函数上的过程。

换句话说,一个“路径操作”就是你定义的一个 API 端点。它由两个核心组成部分构成:

  • 路径(Path):如 /items//users/{user_id} 等 URL 路径;
  • 操作(Operation):如 GET、POST 等 HTTP 方法。

FastAPI 通过装饰器(如 @app.get()@app.post() 等)来声明路径操作。例如:

@app.get("/items/{item_id}")
async def read_item(item_id: int):return {"item_id": item_id}

上面的代码就定义了一个路径为 /items/{item_id},方法为 GET 的路径操作函数 read_item()

FastAPI 会根据路径、方法、函数签名等信息,自动为你生成路由匹配、参数解析、类型校验、自动文档等功能。

因此,“路径操作”是 FastAPI 的核心概念之一,它代表了你 API 的具体业务入口。每一个路径操作函数,都是一个 Web 接口。

路径操作配置

路径操作装饰器支持多种配置参数。

注意:以下参数应直接传递给路径操作装饰器,不能传递给路径操作函数。

status_code 用于定义路径操作响应中的 HTTP 状态码。可以直接传递 int 代码, 比如 404。如果记不住数字码的涵义,也可以用 status 的快捷常量:

from typing import Set, Unionfrom fastapi import FastAPI, status
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):return item

tags 参数的值是由 str 组成的 list (一般只有一个 str ),tags 用于为路径操作添加标签:

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(item: Item):return item@app.get("/items/", tags=["items"])
async def read_items():return [{"name": "Foo", "price": 42}]@app.get("/users/", tags=["users"])
async def read_users():return [{"username": "johndoe"}]

路径装饰器还支持 summary 和 description 这两个参数:

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/",response_model=Item,summary="Create an item",description="Create an item with all the information, name, description, price, tax and a set of unique tags",
)
async def create_item(item: Item):return item

描述内容比较长且占用多行时,可以在函数的 docstring 中声明路径操作的描述,FastAPI 支持从文档字符串中读取描述内容。

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(item: Item):"""Create an item with all the information:- **name**: each item must have a name- **description**: a long description- **price**: required- **tax**: if the item doesn't have tax, you can omit this- **tags**: a set of unique tag strings for this item"""return item

response_description 参数用于定义响应的描述说明:

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/",response_model=Item,summary="Create an item",response_description="The created item",
)
async def create_item(item: Item):"""Create an item with all the information:- **name**: each item must have a name- **description**: a long description- **price**: required- **tax**: if the item doesn't have tax, you can omit this- **tags**: a set of unique tag strings for this item"""return item

注意,response_description 只用于描述响应,description 一般则用于描述路径操作。

deprecated 参数可以把路径操作标记为弃用,无需直接删除:

from fastapi import FastAPIapp = FastAPI()@app.get("/items/", tags=["items"])
async def read_items():return [{"name": "Foo", "price": 42}]@app.get("/users/", tags=["users"])
async def read_users():return [{"username": "johndoe"}]@app.get("/elements/", tags=["items"], deprecated=True)
async def read_elements():return [{"item_id": "Foo"}]

依赖项

FastAPI 提供了简单易用,但功能强大的依赖注入系统。

编程中的「依赖注入」是声明代码(本文中为路径操作函数 )运行所需的,或要使用的「依赖」的一种方式。

然后,由系统(本文中为 FastAPI)负责执行任意需要的逻辑,为代码提供这些依赖(「注入」依赖项)。

依赖注入常用于以下场景:

  • 共享业务逻辑(复用相同的代码逻辑)
  • 共享数据库连接
  • 实现安全、验证、角色权限
  • 等……

上述场景均可以使用依赖注入,将代码重复最小化。

依赖项就是一个函数,且可以使用与路径操作函数相同的参数,与在路径操作函数参数中使用 Body、Query 的方式相同,声明依赖项需要使用 Depends 和一个新的参数:

from typing import Unionfrom fastapi import Depends, FastAPIapp = FastAPI()async def common_parameters(q: Union[str, None] = None, skip: int = 0, limit: int = 100
):return {"q": q, "skip": skip, "limit": limit}@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):return commons@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):return commons

依赖项函数的形式和结构与路径操作函数一样。因此,可以把依赖项当作没有「装饰器」(即,没有 @app.get(“/some-path”) )的路径操作函数。

虽然在路径操作函数的参数中使用 Depends 的方式与 Body、Query 相同,但 Depends 的工作方式略有不同。
这里只能传给 Depends 一个参数。且该参数必须是可调用对象,比如函数。该函数接收的参数和路径操作函数的参数一样。

接收到新的请求时,FastAPI 执行如下操作:

  • 用正确的参数调用依赖项函数(「可依赖项」)
  • 获取函数返回的结果
  • 把函数返回的结果赋值给路径操作函数的参数

在这里插入图片描述
这样,只编写一次代码,FastAPI 就可以为多个路径操作共享这段代码 。

比如,下面有 4 个 API 路径操作(端点):

  • /items/public/
  • /items/private/
  • /users/{user_id}/activate
  • /items/pro/

开发人员可以使用依赖项及其子依赖项为这些路径操作添加不同的权限:

在这里插入图片描述

类作为依赖项

Python 中的 “可调用对象” 是指任何 Python 可以像函数一样 “调用” 的对象。在前面的例子中, 我们从依赖项 (“可依赖对象”) 中返回了一个 dict,我们知道编辑器不能为 dict 提供很多支持(比如补全),因为编辑器不知道 dict 的键和值类型。

因此,在 FastAPI 中,你可以使用一个 Python 类作为一个依赖项。

from fastapi import Depends, FastAPIapp = FastAPI()fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]class CommonQueryParams:def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):self.q = qself.skip = skipself.limit = limit@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):response = {}if commons.q:response.update({"q": commons.q})items = fake_items_db[commons.skip : commons.skip + commons.limit]response.update({"items": items})return response

FastAPI 调用 CommonQueryParams 类。这将创建该类的一个 “实例”,该实例将作为参数 commons 被传递给你的函数。

子依赖项

FastAPI 支持创建含子依赖项的依赖项。并且,可以按需声明任意深度的子依赖项嵌套层级。FastAPI 负责处理解析不同深度的子依赖项。

from typing import Unionfrom fastapi import Cookie, Depends, FastAPIapp = FastAPI()def query_extractor(q: Union[str, None] = None):return qdef query_or_cookie_extractor(q: str = Depends(query_extractor),last_query: Union[str, None] = Cookie(default=None),
):if not q:return last_queryreturn q@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):return {"q_or_cookie": query_or_default}

在这里插入图片描述
如果在同一个路径操作 多次声明了同一个依赖项,例如,多个依赖项共用一个子依赖项,FastAPI 在处理同一请求时,只调用一次该子依赖项。

FastAPI 不会为同一个请求多次调用同一个依赖项,而是把依赖项的返回值进行「缓存」,并把它传递给同一请求中所有需要使用该返回值的「依赖项」。

在高级使用场景中,如果不想使用「缓存」值,而是为需要在同一请求的每一步操作(多次)中都实际调用依赖项,可以把 Depends 的参数 use_cache 的值设置为 False :

async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):return {"fresh_value": fresh_value}

路径操作装饰器依赖项

有时,我们并不需要在路径操作函数中使用依赖项的返回值,但仍要执行或解析该依赖项。

对于这种情况,不必在声明路径操作函数的参数时使用 Depends,而是可以在路径操作装饰器中添加一个由 dependencies 组成的 list。

from fastapi import Depends, FastAPI, Header, HTTPExceptionapp = FastAPI()async def verify_token(x_token: str = Header()):if x_token != "fake-super-secret-token":raise HTTPException(status_code=400, detail="X-Token header invalid")async def verify_key(x_key: str = Header()):if x_key != "fake-super-secret-key":raise HTTPException(status_code=400, detail="X-Key header invalid")return x_key@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():return [{"item": "Foo"}, {"item": "Bar"}]

路径操作装饰器依赖项(以下简称为“路径装饰器依赖项”)的执行或解析方式和普通依赖项一样,但就算这些依赖项会返回值,它们的值也不会传递给路径操作函数。

全局依赖项

有时,我们要为整个应用添加依赖项。

from fastapi import Depends, FastAPI, Header, HTTPExceptionasync def verify_token(x_token: str = Header()):if x_token != "fake-super-secret-token":raise HTTPException(status_code=400, detail="X-Token header invalid")async def verify_key(x_key: str = Header()):if x_key != "fake-super-secret-key":raise HTTPException(status_code=400, detail="X-Key header invalid")return x_keyapp = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])@app.get("/items/")
async def read_items():return [{"item": "Portal Gun"}, {"item": "Plumbus"}]@app.get("/users/")
async def read_users():return [{"username": "Rick"}, {"username": "Morty"}]

使用yield的依赖项

FastAPI支持在完成后执行一些额外步骤的依赖项,为此,你需要使用 yield 而不是 return,然后再编写这些额外的步骤(代码)。

例如,你可以使用这种方式创建一个数据库会话,并在完成后关闭它。

async def get_db():db = DBSession()try:yield dbfinally:db.close()

你可以声明任意数量和层级的树状依赖,而且它们中的任何一个或所有的都可以使用 yield。

FastAPI 会确保每个带有 yield 的依赖中的"退出代码"按正确顺序运行。

例如,dependency_c 可以依赖于 dependency_b,而 dependency_b 则依赖于 dependency_a。

from typing import Annotatedfrom fastapi import Dependsasync def dependency_a():dep_a = generate_dep_a()try:yield dep_afinally:dep_a.close()async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):dep_b = generate_dep_b()try:yield dep_bfinally:dep_b.close(dep_a)async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):dep_c = generate_dep_c()try:yield dep_cfinally:dep_c.close(dep_b)

同样,你可以混合使用带有 yield 或 return 的依赖。你也可以声明一个依赖于多个带有 yield 的依赖,等等。

你可以使用带有 yield 的依赖项,并且可以包含 try 代码块用于捕获异常。同样,你可以在 yield 之后的退出代码中抛出一个 HTTPException 或类似的异常。

from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPExceptionapp = FastAPI()data = {"plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"},"portal-gun": {"description": "Gun to create portals", "owner": "Rick"},
}class OwnerError(Exception):passdef get_username():try:yield "Rick"except OwnerError as e:raise HTTPException(status_code=400, detail=f"Owner error: {e}")@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):if item_id not in data:raise HTTPException(status_code=404, detail="Item not found")item = data[item_id]if item["owner"] != username:raise OwnerError(username)return item

如果你在包含 yield 的依赖项中使用 except 捕获了一个异常,然后你没有重新抛出该异常(或抛出一个新异常),与在普通的Python代码中相同,FastAPI不会注意到发生了异常。

如果你在使用 yield 的依赖项中捕获到了一个异常,你应该再次抛出捕获到的异常,除非你抛出 HTTPException 或类似的其他异常,你可以使用 raise 再次抛出捕获到的异常。

高级依赖项

我们之前看到的所有依赖项都是写死的函数或类,但也可以为依赖项设置参数,避免声明多个不同的函数或类。

Python 可以把类实例变为可调用项。这里说的不是类本身(类本就是可调用项),而是类实例。

为此,需要声明 __call__ 方法,对象加括号再次调用走的是__call__方法!

from fastapi import Depends, FastAPIapp = FastAPI()class FixedContentQueryChecker:def __init__(self, fixed_content: str):self.fixed_content = fixed_contentdef __call__(self, q: str = ""):if q:return self.fixed_content in qreturn Falsechecker = FixedContentQueryChecker("bar")@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):return {"fixed_content_in_query": fixed_content_included}

不要再在 Depends(checker) 中使用 Depends(FixedContentQueryChecker), 而是要使用 checker,因为依赖项是类实例 - checker,不是类。

处理依赖项时,FastAPI 以如下方式调用 checker:

checker(q="somequery")

并用路径操作函数的参数 fixed_content_included 返回依赖项的值。

子路由APIRouter

和Flask的蓝图类型,APIRouter也可以视为一个「迷你 FastAPI」类。

from fastapi import APIRouter, Depends, HTTPExceptionfrom ..dependencies import get_token_headerrouter = APIRouter(prefix="/items",tags=["items"],dependencies=[Depends(get_token_header)],responses={404: {"description": "Not found"}},
)fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}@router.get("/")
async def read_items():return fake_items_db@router.get("/{item_id}")
async def read_item(item_id: str):if item_id not in fake_items_db:raise HTTPException(status_code=404, detail="Item not found")return {"name": fake_items_db[item_id]["name"], "item_id": item_id}@router.put("/{item_id}",tags=["custom"],responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):if item_id != "plumbus":raise HTTPException(status_code=403, detail="You can only update the item: plumbus")return {"item_id": item_id, "name": "The great Plumbus"}
from fastapi import Depends, FastAPIfrom .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, usersapp = FastAPI(dependencies=[Depends(get_query_token)])app.include_router(users.router)
app.include_router(items.router)
app.include_router(admin.router,prefix="/admin",tags=["admin"],dependencies=[Depends(get_token_header)],responses={418: {"description": "I'm a teapot"}},
)@app.get("/")
async def root():return {"message": "Hello Bigger Applications!"}

为什么这样配置?

FastAPI 的路径操作函数中,参数不仅仅用于传递请求数据(如 URL 参数、查询参数、请求体),还可以传递很多其他内容,比如:

  • 请求参数

    • Path, Query, Header, Cookie:显式标记参数来自哪里;
    • 你甚至可以用 Pydantic 模型接收复杂结构(如 JSON 请求体);
  • 依赖注入

    • 使用 Depends() 传入函数或类,做认证、权限控制、数据库连接等;
  • 请求对象

    • Request, Response, BackgroundTasks, UploadFile, Form 等可以直接作为参数传入;
  • 响应控制

    • response_model, status_code, responses 通过装饰器或参数控制返回值格式和状态码;
  • 中间层逻辑

    • 可以将各种逻辑从主业务函数中解耦,写到参数函数中再注入进来。

使用者会感觉,除了业务逻辑,所有功能都“被塞”进了函数参数 —— 这不是错觉,这是 FastAPI 的设计哲学。

FastAPI 的核心设计理念可以总结为以下几点:

✅ 类型提示 + 自动解析:利用 Python 3.6+ 的类型提示(type hinting),FastAPI 能够自动:

  • 校验参数类型(如 int, str | None, list[str]
  • 生成交互式文档
  • 明确函数签名和文档语义

这就意味着你写的每一个参数,既是校验规则,也是接口说明文档,一举多得。

✅ 显式优于隐式(Pythonic):传统框架中很多功能依赖隐式机制,比如:

  • Flask 用全局 request 对象(基于 threading.local);
  • Django 用中间件传递认证信息;

而 FastAPI 强调显式声明:

async def get_current_user(user: User = Depends(authenticate_user)):

谁来、从哪儿来、做什么,都明明白白列出来。

FastAPI 还强调“组合式编程”(Composable Programming):

  • 认证逻辑、数据库连接、缓存控制、分页器等通用逻辑都可以通过参数注入的方式分离出来;
  • 主业务函数只关注业务,其他逻辑通过参数“拼装”进来,提升复用性和测试性。

为什么不采用“传统的请求-响应模型”?

传统模型(如 Flask)通常是:

from flask import request@app.route("/user")
def get_user():user_id = request.args.get("id")return {"user_id": user_id}

它有几个问题:

  • 参数缺乏类型校验;
  • 请求来源混乱(你不知道 id 是 query 还是 path);
  • 逻辑与框架绑定紧密(全局对象如 request 不易测试);
  • 很难自动生成高质量的 API 文档。

FastAPI 的方式虽然一开始看着“重”,但具备:

  • 更强的结构化
  • 更好的 IDE 补全和类型检查
  • 自动文档
  • 解耦逻辑的能力
  • 明确的请求来源

说到底,FastApi的目标是构建 可维护、自动文档化、类型安全、可测试的 Web API,而不是一个高可用的Web API。

中间件

你可以向 FastAPI 应用添加中间件,"中间件"是一个函数,它在每个请求被特定的路径操作处理之前,以及在每个响应返回之前工作.

  • 它接收你的应用程序的每一个请求.
  • 然后它可以对这个请求做一些事情或者执行任何需要的代码.
  • 然后它将请求传递给应用程序的其他部分 (通过某种路径操作).
  • 然后它获取应用程序生产的响应 (通过某种路径操作).
  • 它可以对该响应做些什么或者执行任何需要的代码.
  • 然后它返回这个 响应.

如果你使用了 yield 关键字依赖, 依赖中的退出代码将在执行中间件后执行,可以将yield理解为go的defer。

要创建中间件,你可以在函数的顶部使用装饰器 @app.middleware("http")

中间件的参数包括:

  • request:传入的请求对象;
  • 一个函数 call_next,它接收 request 作为参数;
    • call_next(request) 会将请求传递给对应的路径操作函数;
    • 然后返回由路径操作生成的 response
  • 你可以在返回 response 之前对其进一步修改。
import timefrom fastapi import FastAPI, Requestapp = FastAPI()@app.middleware("http")
async def add_process_time_header(request: Request, call_next):start_time = time.perf_counter()response = await call_next(request)process_time = time.perf_counter() - start_timeresponse.headers["X-Process-Time"] = str(process_time)return response

CORS(跨域资源共享)

CORS 或者「跨域资源共享」 指浏览器中运行的前端拥有与后端通信的 JavaScript 代码,而后端处于与前端不同的「源」的情况。

源是协议(http、https)、域(myapp.com、localhost、localhost.tiangolo.com)以及端口(80、443、8080)的组合。

因此,以下都是不同的源

  • http://localhost
  • https://localhost
  • http://localhost:8080

即使它们都在 localhost 中,但由于使用了不同的协议或端口,它们仍然被视为不同的「源」。

假设你的浏览器中有一个前端运行在 http://localhost:8080,并且它的 JavaScript 正在尝试与运行在 http://localhost 的后端通信(因为我们没有指定端口,浏览器会采用默认的端口 80)。

然后,浏览器会向后端发送一个 HTTP OPTIONS 请求。如果后端发送了适当的 headers 来授权来自这个不同源(http://localhost:8080)的通信,浏览器将允许前端的 JavaScript 向后端发送请求。

为此,后端必须有一个「允许的源」列表,在这种情况下,它必须包含 http://localhost:8080,前端才能正常工作。

可以使用 "*"(一个「通配符」)声明这个列表,表示全部都是允许的。

但这仅允许某些类型的通信,不包括所有涉及凭据的内容,如 Cookies 以及那些使用 Bearer 令牌的授权 headers 等。

因此,为了一切都能正常工作,最好显式地指定允许的源,你可以在 FastAPI 应用中使用 CORSMiddleware 来配置它。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()origins = ["http://localhost.tiangolo.com","https://localhost.tiangolo.com","http://localhost","http://localhost:8080",
]app.add_middleware(CORSMiddleware,allow_origins=origins,allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)@app.get("/")
async def main():return {"message": "Hello World"}

默认情况下,这个 CORSMiddleware 实现所使用的默认参数较为保守,所以你需要显式地启用特定的源、方法或 headers,以便浏览器能够在跨域上下文中使用它们。

支持的参数如下:

  • allow_origins
    一个允许跨域请求的源列表。例如:
    ['https://example.org', 'https://www.example.org']
    你也可以使用 ['*'] 来允许任何源。

  • allow_origin_regex
    一个正则表达式字符串,匹配的源将允许跨域请求。
    例如:'https://.*\.example\.org'

  • allow_methods
    一个允许跨域请求的 HTTP 方法列表。默认为 ['GET']
    可以使用 ['*'] 来允许所有标准方法(如 POSTPUTDELETE 等)。

  • allow_headers
    一个允许跨域请求的 HTTP 请求头列表。默认为 []
    使用 ['*'] 可以允许所有的请求头。
    其中 AcceptAccept-LanguageContent-LanguageContent-Type 总是被允许的。

  • allow_credentials
    是否允许携带 cookies(凭证)。默认为 False
    注意:若设置为 True,则 allow_origins 不能为 ['*'],必须明确指定源。

  • expose_headers
    指示哪些响应头可以被浏览器访问。默认为 []

  • max_age
    指定浏览器缓存预检请求(CORS 响应)的最长时间(单位:秒)。默认为 600

CORSMiddleware 主要响应两种特定类型的 HTTP 请求:

  1. 预检请求(OPTIONS):浏览器在实际请求前自动发起,用于验证是否允许跨域;
  2. 实际跨域请求:在通过预检后,才允许真正的 API 请求被发送。

其他中间件

FastAPI(实际上是 Starlette)提供了一种更简单的方式,能让内部中间件在处理服务器错误的同时,还能让自定义异常处理器正常运作。app.add_middleware() 的第一个参数是中间件的类,其它参数则是要传递给中间件的参数。

HTTPSRedirectMiddleware

HTTPSRedirectMiddleware 强制所有传入请求必须是 httpswss
任何传向 httpws 的请求都会被自动重定向至安全方案。

示例:

from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddlewareapp = FastAPI()
app.add_middleware(HTTPSRedirectMiddleware)@app.get("/")
async def main():return {"message": "Hello World"}

TrustedHostMiddleware

TrustedHostMiddleware 强制所有传入请求必须正确设置 Host 请求头,以防止 HTTP 主机头攻击。

示例:

from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddlewareapp = FastAPI()
app.add_middleware(TrustedHostMiddleware,allowed_hosts=["example.com", "*.example.com"]
)@app.get("/")
async def main():return {"message": "Hello World"}

支持参数:

  • allowed_hosts
    允许的域名(主机名)列表。例如:*.example.com 可匹配任意子域名。
    也可使用 ["*"] 允许任意主机名,或不添加中间件跳过验证。
    如果传入请求未通过验证,将返回 400 响应。

GZipMiddleware

GZipMiddleware 用于在客户端的 Accept-Encoding 请求头包含 gzip 时,自动返回 GZip 压缩响应。
它可处理标准响应与流响应。

示例:

from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddlewareapp = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=5)@app.get("/")
async def main():return "somebigcontent"

支持参数:

  • minimum_size
    响应内容小于该字节数时不进行 GZip 压缩。默认值为 500 字节。

  • compresslevel
    GZip 压缩级别,取值范围为 0-9,默认为 9(压缩率最高,速度最慢)。

和依赖项的区别

特性中间件(Middleware)依赖项(Dependencies)
作用层级应用于整个应用的请求/响应生命周期,处理所有请求作用于特定路径操作函数或多个函数,支持参数注入
执行时机请求进入 FastAPI 后最先执行,响应返回前最后执行在路径操作函数执行前运行,支持参数准备、验证、注入
功能范围通常用于处理跨域、认证、日志、限流、压缩等全局功能用于业务逻辑中参数处理、复用代码、复杂校验、依赖注入
返回值返回修改后的响应(Response)返回参数值,作为路径操作函数参数
定义方式使用 @app.middleware("http") 装饰函数或 add_middleware()使用函数或类,并通过 Depends() 注入
设计目标处理所有请求的公共横切关注点解耦业务逻辑,复用代码,提高代码模块化
例子日志记录、中间件认证、请求限流、压缩响应验证用户身份、获取数据库连接、复用公共参数

为什么 FastAPI 分离中间件和依赖? FastAPI 的设计追求高度的灵活性和清晰的职责分离:

  • 中间件 设计为应用级的请求/响应处理链,用于统一处理所有请求,不依赖具体业务逻辑。
  • 依赖项 设计为路径操作函数的参数注入机制,便于将业务相关的功能按需引入,支持复杂的参数验证与逻辑复用。

这种设计让开发者可以:

  • 在不修改业务代码的情况下,统一处理请求和响应(通过中间件);
  • 在业务代码中灵活注入所需资源或逻辑(通过依赖),保持代码简洁且模块化。

后台任务

你可以定义在返回响应后运行的后台任务,这对需要在请求之后执行的操作很有用,但客户端不必在接收响应之前等待操作完成。

首先导入 BackgroundTasks 并在 路径操作函数 中使用类型声明 BackgroundTasks 定义一个参数:

from fastapi import BackgroundTasks, FastAPIapp = FastAPI()def write_notification(email: str, message=""):with open("log.txt", mode="w") as email_file:content = f"notification for {email}: {message}"email_file.write(content)@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):background_tasks.add_task(write_notification, email, message="some notification")return {"message": "Notification sent in the background"}

FastAPI 会创建一个 BackgroundTasks 类型的对象并作为该参数传入,在你的 路径操作函数 里,用 .add_task() 方法将任务函数传到 后台任务 对象中!

.add_task() 接收以下参数:

  • 在后台运行的任务函数(write_notification)。
  • 应按顺序传递给任务函数的任意参数序列(email)。
  • 应传递给任务函数的任意关键字参数(message=“some notification”)。

使用 BackgroundTasks 也适用于依赖注入系统,你可以在多个级别声明 BackgroundTasks 类型的参数:在 路径操作函数 里,在依赖中(可依赖),在子依赖中,等等。

FastAPI 知道在每种情况下该做什么以及如何复用同一对象,因此所有后台任务被合并在一起并且随后在后台运行:

from typing import Annotatedfrom fastapi import BackgroundTasks, Depends, FastAPIapp = FastAPI()def write_log(message: str):with open("log.txt", mode="a") as log:log.write(message)def get_query(background_tasks: BackgroundTasks, q: str | None = None):if q:message = f"found query: {q}\n"background_tasks.add_task(write_log, message)return q@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):message = f"message to {email}\n"background_tasks.add_task(write_log, message)return {"message": "Message sent"}

实际上就是用线程池去执行这些任务,只不过是这个线程池是基于threading + anyio + asyncio的牛逼的线程池,比我们自己创建的pool = ThreadingPool()高级一些。

生命周期事件

你可以定义在应用启动前执行的逻辑(代码)。这意味着在应用开始接收请求之前,这些代码只会被执行一次。

同样地,你可以定义在应用关闭时应执行的逻辑。在这种情况下,这段代码将在处理可能的多次请求后执行一次。

因为这段代码在应用开始接收请求之前执行,也会在处理可能的若干请求之后执行,它覆盖了整个应用程序的生命周期("生命周期"这个词很重要😉)。

假设你有几个机器学习的模型,你想要用它们来处理请求。相同的模型在请求之间是共享的,因此并非每个请求或每个用户各自拥有一个模型。假设加载模型可能需要相当长的时间,因为它必须从磁盘读取大量数据。因此你不希望每个请求都加载它。

你可以在模块/文件的顶部加载它,但这也意味着即使你只是在运行一个简单的自动化测试,它也会加载模型,这样测试将变慢,因为它必须在能够独立运行代码的其他部分之前等待模型加载完成。

这就是我们要解决的问题——在处理请求前加载模型,但只是在应用开始接收请求前,而不是代码执行时。

你可以使用FastAPI()应用的lifespan参数和一个上下文管理器来定义启动和关闭的逻辑,FastAPI() 的 lifespan 参数接受一个异步上下文管理器,所以我们可以把我们定义的上下文管理器 lifespan 传给它。

from contextlib import asynccontextmanagerfrom fastapi import FastAPIdef fake_answer_to_everything_ml_model(x: float):return x * 42ml_models = {}@asynccontextmanager
async def lifespan(app: FastAPI):# Load the ML modelml_models["answer_to_everything"] = fake_answer_to_everything_ml_modelyield# Clean up the ML models and release the resourcesml_models.clear()app = FastAPI(lifespan=lifespan)@app.get("/predict")
async def predict(x: float):result = ml_models["answer_to_everything"](x)return {"result": result}

你可以像上面一样创建了一个上下文管理器或者异步上下文管理器,它的作用是在进入 with 块时,执行 yield 之前的代码,并且在离开 with 块时,执行 yield 后面的代码。

配置启动和关闭事件的推荐方法是使用 FastAPI() 应用的 lifespan 参数,如前所示。如果你提供了一个 lifespan 参数,启动(startup)和关闭(shutdown)事件处理器将不再生效。要么使用 lifespan,要么配置所有事件,两者不能共用。

使用 startup 事件声明 app 启动前运行的函数:

from fastapi import FastAPIapp = FastAPI()items = {}@app.on_event("startup")
async def startup_event():items["foo"] = {"name": "Fighters"}items["bar"] = {"name": "Tenders"}@app.get("/items/{item_id}")
async def read_items(item_id: str):return items[item_id]

使用 shutdown 事件声明 app 关闭时运行的函数:

from fastapi import FastAPIapp = FastAPI()@app.on_event("shutdown")
def shutdown_event():with open("log.txt", mode="a") as log:log.write("Application shutdown")@app.get("/items/")
async def read_items():return [{"name": "Foo"}]

启动和关闭的逻辑很可能是连接在一起的,你可能希望启动某个东西然后结束它,获取一个资源然后释放它等等。

在不共享逻辑或变量的不同函数中处理这些逻辑比较困难,因为你需要在全局变量中存储值或使用类似的方式。

SQL(关系型)数据库——现代化ORM

FastAPI 并不要求您使用 SQL(关系型)数据库。您可以使用任何想用的数据库。

SQLModel 是基于 SQLAlchemy 和 Pydantic 构建的。它由 FastAPI 的同一作者制作,旨在完美匹配需要使用 SQL 数据库的 FastAPI 应用程序。

由于 SQLModel 基于 SQLAlchemy,因此您可以轻松使用任何由 SQLAlchemy 支持的数据库(这也让它们被 SQLModel 支持),例如:

  • PostgreSQL
  • MySQL
  • SQLite
  • Oracle
  • Microsoft SQL Server 等.

安装SQLModel:pip install sqlmodel

导入 SQLModel 并创建一个数据库模型:

from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass Hero(SQLModel, table=True):id: int | None = Field(default=None, primary_key=True)name: str = Field(index=True)age: int | None = Field(default=None, index=True)secret_name: str# Code below omitted 👇

Hero 类与 Pydantic 模型非常相似(实际上,从底层来看,它确实就是一个 Pydantic 模型)。

有一些区别:

  • table=True 会告诉 SQLModel 这是一个表模型,它应该表示 SQL 数据库中的一个表,而不仅仅是一个数据模型(就像其他常规的 Pydantic 类一样)。

  • Field(primary_key=True) 会告诉 SQLModel id 是 SQL 数据库中的主键(您可以在 SQLModel 文档中了解更多关于 SQL 主键的信息)。

  • 把类型设置为 int | None,SQLModel 就能知道该列在 SQL 数据库中应该是 INTEGER 类型,并且应该是 NULLABLE

  • Field(index=True) 会告诉 SQLModel 应该为此列创建一个 SQL 索引,这样在读取按此列过滤的数据时,程序能在数据库中进行更快的查找。

  • SQLModel 会知道声明为 str 的内容将是类型为 TEXT(或 VARCHAR,具体取决于数据库)的 SQL 列。

SQLModel 的引擎 engine(实际上它是一个 SQLAlchemy engine )是用来与数据库保持连接的。您只需构建一个 engine,来让您的所有代码连接到同一个数据库。

# Code above omitted 👆sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)# Code below omitted 👇

使用 check_same_thread=False 可以让 FastAPI 在不同线程中使用同一个 SQLite 数据库。这很有必要,因为单个请求可能会使用多个线程(例如在依赖项中)。

然后,我们来添加一个函数,使用 SQLModel.metadata.create_all(engine) 为所有表模型创建表。

# Code above omitted 👆def create_db_and_tables():SQLModel.metadata.create_all(engine)# Code below omitted 👇

Session 会存储内存中的对象并跟踪数据中所需更改的内容,然后它使用 engine 与数据库进行通信。

我们会使用 yield 创建一个 FastAPI 依赖项,为每个请求提供一个新的 Session 。这确保我们每个请求使用一个单独的会话。

# Code above omitted 👆def get_session():with Session(engine) as session:yield sessionSessionDep = Annotated[Session, Depends(get_session)]# Code below omitted 👇

我们会在应用程序启动时创建数据库表。

# Code above omitted 👆app = FastAPI()@app.on_event("startup")
def on_startup():create_db_and_tables()# Code below omitted 👇

这里,我们使用 SessionDep 依赖项(一个 Session )将新的 Hero 添加到 Session 实例中,提交更改到数据库,刷新 hero 中的数据,并返回它。

# Code above omitted 👆@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:session.add(hero)session.commit()session.refresh(hero)return hero# Code below omitted 👇

我们可以使用 select() 从数据库中读取 Hero 类,并利用 limit 和 offset 来对结果进行分页。

# Code above omitted 👆@app.get("/heroes/")
def read_heroes(session: SessionDep,offset: int = 0,limit: Annotated[int, Query(le=100)] = 100,
) -> list[Hero]:heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes# Code below omitted 👇

我们可以读取单个 Hero 。

# Code above omitted 👆@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep) -> Hero:hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")return hero# Code below omitted 👇

我们也可以删除单个 Hero 。

# Code above omitted 👆@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")session.delete(hero)session.commit()return {"ok": True}

静态文件

您可以使用 StaticFiles从目录中自动提供静态文件。

  • 导入StaticFiles。
  • “挂载”(Mount) 一个 StaticFiles() 实例到一个指定路径。
from fastapi import FastAPI
from fastapi.staticfiles import StaticFilesapp = FastAPI()app.mount("/static", StaticFiles(directory="static"), name="static")

“挂载” 表示在特定路径添加一个完全"独立的"应用,然后负责处理所有子路径。

这个 “子应用” 会被 “挂载” 到第一个 “/static” 指向的子路径。因此,任何以"/static"开头的路径都会被它处理。

  • directory=“static” 指向包含你的静态文件的目录名字。
  • name=“static” 提供了一个能被FastAPI内部使用的名字。

模板

Flask 等工具使用的 Jinja2 是最用的模板引擎。(pip install jinja2)

使用 Jinja2Templates

  • 导入 Jinja2Templates
  • 创建 可复用的 templates 对象
  • 在返回模板的路径操作中 声明 Request 参数
  • 使用 templates 渲染并返回 TemplateResponse
    • 传递模板的名称
    • 传递 request 对象
    • 传递一个包含多个键值对(用于 Jinja2 模板)的 "context" 字典
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templatesapp = FastAPI()app.mount("/static", StaticFiles(directory="static"), name="static")templates = Jinja2Templates(directory="templates")@app.get("/items/{id}", response_class=HTMLResponse)
async def read_item(request: Request, id: str):return templates.TemplateResponse(request=request, name="item.html", context={"id": id})

通过声明 response_class=HTMLResponse,API 文档就能识别响应的对象是 HTML。

编写模板 templates/item.html,代码如下:

<html>
<head><title>Item Details</title><link href="{{ url_for('static', path='/styles.css') }}" rel="stylesheet">
</head>
<body><h1><a href="{{ url_for('read_item', id=id) }}">Item ID: {{ id }}</a></h1>
</body>
</html>

部署

部署 FastAPI 应用程序相对容易。

简而言之,使用 fastapi run 来运行您的 FastAPI 应用程序:

在这里插入图片描述
这在大多数情况下都能正常运行。😎

FastAPI 使用了一种用于构建 Python Web 框架和服务器的标准,称为 ASGI。FastAPI 本质上是一个 ASGI Web 框架

要在远程服务器上运行 FastAPI 应用(或任何其他 ASGI 应用),您需要一个 ASGI 服务器程序,例如 Uvicorn。它是 fastapi 命令默认使用的 ASGI 服务器。

除此之外,还有其他一些可选的 ASGI 服务器,例如:

  • Uvicorn:高性能 ASGI 服务器。
  • Hypercorn:与 HTTP/2 和 Trio 等兼容的 ASGI 服务器。
  • Daphne:为 Django Channels 构建的 ASGI 服务器。
  • Granian:基于 Rust 的 HTTP 服务器,专为 Python 应用设计。
  • NGINX Unit:轻量级且灵活的 Web 应用运行时环境。

当您安装 FastAPI 时,它自带一个生产环境服务器——Uvicorn,并且您可以使用 fastapi run 命令来启动它。

不过,您也可以手动安装 ASGI 服务器:pip install "uvicorn[standard]"

如果您手动安装了 ASGI 服务器,通常需要以特定格式传递一个导入字符串,以便服务器能够正确导入您的 FastAPI 应用:

uvicorn main:app --host 0.0.0.0 --port 80

到目前为止,在文档中的所有教程中,您可能一直是在运行一个服务器程序,例如使用 fastapi 命令来启动 Uvicorn,而它默认运行的是单进程模式。

部署应用程序时,您可能希望进行一些进程复制,以利用多核 CPU 并能够处理更多请求。

您可以使用 --workers 命令行选项来启动多个工作进程:

在这里插入图片描述

在这里插入图片描述
如果您的代码加载 1 GB 大小的机器学习模型,则当您使用 API 运行一个进程时,它将至少消耗 1 GB RAM。 如果您启动 4 个进程(4 个工作进程),每个进程将消耗 1 GB RAM。 因此,您的 API 总共将消耗 4 GB RAM。

如果您的远程服务器或虚拟机只有 3 GB RAM,尝试加载超过 4 GB RAM 将导致问题。

在这里插入图片描述

相关文章:

  • ubuntu安装vmware启动虚拟机失败
  • 计算机组成与体系结构:缓存一致性(Cache Coherence)
  • 第四章:WebSocket 通信机制全解与客户端发包实录
  • BOM知识点
  • 线程池学习笔记
  • Linux系统编程——system函数和popen函数的使用方法以及区别
  • 三维重建(二十三)——各种参数的测试(废案)
  • IIS入门指南:原理、部署与实战
  • 容器化:用于机器学习的 Docker 和 Kubernetes
  • SRIO(Serial RapidIO)握手流程
  • 算符优先分析之输入串分析
  • 【深度学习】残差网络(ResNet)
  • python学习day29
  • Neovim 如何安装和配置缩进标识插件 indent-blankline.nvim
  • 贝塞尔曲线原理
  • Manus vs Lovart:AI Agent技术深度解析与实战指南
  • MyBatis框架(入门)
  • Arduino UNO采集血压/心率/血氧/体温等数据传输到Blinker云平台 手机APP显示数据
  • 【深度学习基础】从感知机到多层神经网络:模型原理、结构与计算过程全解析
  • 计算机系统---TPU(张量处理单元)
  • 4年间职务侵占、受贿逾亿元,北京高院:严惩民企内部腐败
  • 女巫的继承者们
  • 消费维权周报丨上周涉汽车类投诉较多,涉加油“跳枪”等问题
  • 美国前总统拜登确诊前列腺癌
  • 中国田径巡回赛西安站完赛:男子跳远石雨豪夺冠,女子跳高刘肼毅折桂
  • 南京艺术学院博导、雕塑家尹悟铭病逝,年仅45岁