FastAPI 快速指南

---

---

FastAPI - 简介

FastAPI 是一个现代化的 Python Web 框架，非常高效地构建 API。它基于 Python 自 3.6 版本开始添加的类型提示功能。它是 Python 最快的 Web 框架之一。

• 由于它基于 Starlette 和 Pydantic 库的功能，其性能处于最佳水平，与 NodeJS 和 Go 相当。

• 除了提供高性能外，FastAPI 还提供了显著的开发速度，减少了代码中人为错误，易于学习，并且完全可以用于生产环境。

• FastAPI 完全兼容众所周知的 API 标准，即 OpenAPI 和 JSON Schema。

FastAPI 由 Sebastian Ramirez 于 2018 年 12 月开发。目前可用的版本为 FastAPI 0.68.0。

FastAPI – 环境设置

要安装 FastAPI（最好在虚拟环境中），请使用 pip 安装程序。

pip3 install fastapi

FastAPI 依赖于 Starlette 和 Pydantic 库，因此它们也会被安装。

使用 PIP 安装 Uvicorn

FastAPI 没有自带任何内置的服务器应用程序。要运行 FastAPI 应用程序，您需要一个名为 uvicorn 的 ASGI 服务器，因此也请使用 pip 安装程序安装它。它还会安装 uvicorn 的依赖项 - asgiref、click、h11 和 typing-extensions

pip3 install uvicorn

安装这两个库后，我们可以检查到目前为止安装的所有库。

pip3 freeze
asgiref==3.4.1
click==8.0.1
colorama==0.4.4
fastapi==0.68.0
h11==0.12.0
importlib-metadata==4.6.4
pydantic==1.8.2
starlette==0.14.2
typing-extensions==3.10.0.0
uvicorn==0.15.0
zipp==3.5.0

FastAPI - Hello World

入门

创建 FastAPI 应用程序的第一步是声明 FastAPI 类的应用程序对象。

from fastapi import FastAPI
app = FastAPI()

此 app 对象是应用程序与客户端浏览器交互的主要点。uvicorn 服务器使用此对象来监听客户端的请求。

下一步是创建路径操作。路径是一个 URL，当客户端访问该 URL 时，会调用映射到该 URL 的其中一个 HTTP 方法，并执行关联的函数。我们需要将视图函数绑定到 URL 和相应的 HTTP 方法。例如，index() 函数对应于具有 ‘get’ 操作的 ‘/’ 路径。

@app.get("/")
async def root():
   return {"message": "Hello World"}

该函数返回一个 JSON 响应，但是，它也可以返回 dict、list、str、int 等。它还可以返回 Pydantic 模型。

将以下代码保存为 main.py

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

通过提及实例化 FastAPI 应用程序对象的代码文件来启动 uvicorn 服务器。

uvicorn main:app --reload
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [28720]
INFO: Started server process [28722]
INFO: Waiting for application startup.
INFO: Application startup complete.

打开浏览器并访问 https://127.0.0.1:/8000。您将在浏览器窗口中看到 JSON 响应。

FastAPI - OpenAPI

在浏览器中输入以下 URL 以自动生成交互式文档。

http://127.0.0.1:8000/docs

FastAPI 使用 Swagger UI 生成此文档。浏览器将显示以下内容 -

单击 '试一下' 按钮，然后单击随后出现的 '执行' 按钮。

您可以看到内部执行的 Curl 命令、请求 URL、响应头以及服务器响应的 JSON 格式。

FastAPI 使用 OpenAPI 规范生成模式。该规范确定如何定义 API 路径、路径参数等。OpenAPI 标准定义的 API 模式决定了如何使用 JSON Schema 发送数据。从浏览器访问 http://127.0.0.1:8000/openapi.json。将显示如下格式整齐的 JSON 响应 -

{
   "openapi": "3.0.2",
   "info": {
      "title": "FastAPI",
      "version": "0.1.0"
   },
   "paths": {
      "/": {
         "get": {
            "summary": "Index",
            "operationId": "index__get",
            "responses": {
               "200": {
                  "description": "Successful Response",
                  "content": {
                     "application/json": {
                        "schema": {}
                     }
                  }
               }
            }
         }
      }
   }
}

FastAPI 还支持 Redoc 提供的另一种自动文档方法 ( https://github.com/Redocly/redoc)。

在浏览器的地址栏中输入 https://127.0.0.1:8000/redoc 作为 URL。

FastAPI - Uvicorn

与 Flask 框架不同，FastAPI 不包含任何内置的开发服务器。因此我们需要 Uvicorn。它实现了 ASGI 标准并且速度极快。ASGI 代表 异步服务器网关接口。

符合 WSGI（Web 服务器网关接口 - 旧标准）的 Web 服务器不适合 asyncio 应用程序。实现 ASGI 规范的 Python Web 框架（如 FastAPI）提供高速性能，与使用 Node 和 Go 构建的 Web 应用程序相当。

Uvicorn 使用 uvloop 和 httptools 库。它还提供对 HTTP/2 和 WebSockets 的支持，而 WSGI 无法处理这些协议。uvloop 类似于内置的 asyncio 事件循环。httptools 库处理 http 协议。

如前所述，安装 Uvicorn 将以最少的依赖项安装它。但是，标准安装还将安装基于 cython 的依赖项以及其他一些库。

pip3 install uvicorn(standard)

这样，将支持 WebSockets 协议。此外，将安装 PyYAML 以允许您提供 .yaml 文件。

如前所述，应用程序在 Uvicorn 服务器上使用以下命令启动 -

uvicorn main:app –reload

--reload 选项启用调试模式，以便 app.py 中的任何更改都会自动反映出来，并且客户端浏览器上的显示也会自动刷新。此外，可以使用以下命令行选项 -

除了从命令行启动 Uvicorn 服务器之外，还可以以编程方式启动它。

示例

在 Python 代码中，使用上面列出的任何参数调用 uvicorn.run() 方法 -

import uvicorn
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def index():
   return {"message": "Hello World"}
if __name__ == "__main__":
   uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

现在，按如下方式运行此 app.py 作为 Python 脚本 -

(fastapienv) C:\fastapienv>python app.py

因此，Uvicorn 服务器将在调试模式下启动。

FastAPI - 类型提示

FastAPI 广泛使用 Python 3.5 及更高版本中提供的 类型提示功能。事实上，Python 以动态类型语言而闻名。它恰好也是 Python 的独特功能。在 Python 代码中，无需声明变量属于某种类型，其类型由动态分配给它的瞬时值确定。Python 的解释器不执行类型检查，因此容易出现运行时异常。

在以下示例中，定义了一个 division() 函数，它有两个参数并返回它们的除法结果，假设这些参数将是数字。

>>> def division(a, b):
   return a/b
>>> division(10, 4)
2.5
>>> division(10, 2.5)
4.0

但是，如果传递给函数的值之一恰好是非数字，则会导致 TypeError，如下所示 -

>>> division("Python",5)
TypeError: unsupported operand type(s) for /: 'str' and 'int'

即使是像 IDLE 这样的基本编码环境也会指示该函数需要两个参数，但不会指定类型，因为它们尚未声明。

Python 的新的类型提示功能有助于提示用户传递的参数的预期类型。这是通过在参数之后添加冒号和数据类型来完成的。我们将重新定义 division() 函数，如下所示 -

请注意，在调用函数时，Python 会提示传递的每个参数的预期类型。但是，如果传递了不兼容的值，这并不能阻止 TypeError 出现。您必须使用静态类型检查器（如 MyPy）在运行前检查兼容性。

就像函数定义中的形式参数一样，可以为函数的返回值提供类型提示。只需在函数定义语句中的冒号符号之前（函数块开始之后）添加箭头（->）和类型即可。

但是，如前所述，如果传递给函数的值不兼容，或者函数返回的值不兼容，Python 会报告 TypeError。可以使用 MyPy 静态类型检查器检测此类错误。首先安装 mypy 包。

pip3 install mypy

将以下代码保存为 typecheck.py

def division(x:int, y:int) -> int:
   return (x//y)
a=division(10,2)
print (a)
b=division(5,2.5)
print (b)
c=division("Hello",10)
print (c)

使用 mypy 检查此代码是否存在类型错误。

C:\python37>mypy typechk.py
typechk.py:7: error: Argument 2 to "division" has incompatible
type "float"; expected "int"
typechk.py:10: error: Argument 1 to "division" has
incompatible type "str"; expected "int"
Found 2 errors in 1 file (checked 1 source file)

函数的第二个和第三个调用中存在错误。在第二个调用中，传递给 y 的值是 float，而预期的是 int。在第三个调用中，传递给 x 的值是 str，而预期的是 int。（请注意，// 运算符返回整数除法）

所有标准数据类型都可以用作类型提示。这可以在全局变量、作为函数参数的变量、函数定义内部等中完成。

x: int = 3
y: float = 3.14
nm: str = 'abc'
married: bool = False
names: list = ['a', 'b', 'c']
marks: tuple = (10, 20, 30)
marklist: dict = {'a': 10, 'b': 20, 'c': 30}

Python 版本 3.5 及更高版本标准库中的新增功能是 typing 模块。它为相应的标准集合类型定义了特殊类型。typing 模块上的类型是 List、Tuple、Dict 和 Sequence。它还包含 Union 和 Optional 类型。请注意，数据类型的标准名称全部为小写，而 typing 模块中的名称第一个字母为大写。使用此功能，我们可以要求特定类型的集合。

from typing import List, Tuple, Dict
# following line declares a List object of strings.
# If violated, mypy shows error
cities: List[str] = ['Mumbai', 'Delhi', 'Chennai']
# This is Tuple with three elements respectively
# of str, int and float type)
employee: Tuple[str, int, float] = ('Ravi', 25, 35000)
# Similarly in the following Dict, the object key should be str
# and value should be of int type, failing which
# static type checker throws error
marklist: Dict[str, int] = {'Ravi': 61, 'Anil': 72}

FastAPI - IDE 支持

Python 的类型提示功能在几乎所有 IDE（集成开发环境），如 PyCharm 和 VS Code 中都得到了最有效的利用，以提供动态自动完成功能。

让我们看看 VS Code 如何使用类型提示在编写代码时提供自动完成建议。在下面的示例中，定义了一个名为 sayhello 的函数，其中 name 作为参数。该函数通过在 name 参数之间添加空格将“Hello”连接到 name 参数，从而返回一个字符串。此外，还需要确保 name 的第一个字母为大写。

Python 的 str 类具有用于此目的的 capitalize() 方法，但如果在键入代码时不记得它，则必须在其他地方搜索它。如果在 name 之后添加一个点，您期望看到属性列表，但什么也没有显示，因为 Python 不知道 name 变量的运行时类型是什么。

这里，类型提示派上用场了。在函数定义中将字符串作为名称的类型包含在内。现在，当您在 name 后按点 (.) 时，会显示所有字符串方法的下拉列表，从中可以选择所需的方法（在本例中为 capitalize()）。

也可以将类型提示与用户定义的类一起使用。在下面的示例中，定义了一个矩形类，并为__init__()构造函数的参数提供了类型提示。

class rectangle:
   def __init__(self, w:int, h:int) ->None:
      self.width=w
      self.height=h

下面是一个使用上述矩形类对象作为参数的函数。声明中使用的类型提示是类的名称。

def area(r:rectangle)->int:
   return r.width*r.height
r1=rectangle(10,20)
print ("area = ", area(r1))

在这种情况下，IDE 编辑器也提供了自动完成支持，提示实例属性列表。下面是PyCharm编辑器的屏幕截图。

FastAPI广泛使用类型提示。此功能随处可见，例如路径参数、查询参数、标头、主体、依赖项等，以及验证来自传入请求的数据。OpenAPI 文档生成也使用类型提示。

FastAPI - REST 架构

表述性状态转移(REST) 是一种软件架构风格。REST 定义了 Web 应用程序的架构应该如何运作。它是一种基于资源的架构，其中 REST 服务器托管的所有内容（文件、图像或数据库表中的行）都是资源，具有多种表示形式。

REST 建议某些架构约束。

• 统一接口

• 无状态性

• 客户端-服务器

• 可缓存性

• 分层系统

• 按需代码

REST 约束具有以下优势：

• 可扩展性

• 简单性

• 可修改性

• 可靠性

• 可移植性

• 可见性

REST 使用 HTTP 动词或方法对资源进行操作。POST、GET、PUT 和 DELETE 方法分别执行 CREATE、READ、UPDATE 和 DELETE 操作。

FastAPI - 路径参数

现代 Web 框架使用路由或端点作为 URL 的一部分，而不是基于文件的 URL。这有助于用户更有效地记住应用程序 URL。在 FastAPI 中，它被称为路径。路径或路由是第一个“/”之后 URL 的一部分。

例如，在以下 URL 中：

https://127.0.0.1:8000/hello/TutorialsPoint

路径或路由将是

/hello/TutorialsPoint

在 FastAPI 中，这样的路径字符串作为参数传递给操作装饰器。这里的操作是指浏览器用于发送数据的 HTTP 动词。这些操作包括 GET、PUT 等。操作装饰器（例如，@app.get("/")）紧跟一个函数，当访问指定的 URL 时执行该函数。在下面的示例中：

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

这里，("/") 是路径，get 是操作，@app.get("/") 是路径操作装饰器，它下面的 index() 函数称为路径操作函数。

任何以下 HTTP 动词都可以用作操作。

函数定义中的async关键字告诉 FastAPI 它应该异步运行，即不阻塞当前执行线程。但是，也可以在没有 async 前缀的情况下定义路径操作函数。

此装饰函数返回 JSON 响应。虽然它可以返回几乎任何 Python 对象，但它会自动转换为 JSON。在本教程的后面，我们将看到这样的函数如何返回Pydantic模型对象。

URL 的端点或路径可以有一个或多个可变参数。它们可以通过使用 Python 的字符串格式化表示法来接受。在上面的示例 URL https://127.0.0.1:8000/hello/TutorialsPoint 中，最后一个值可能会在每个客户端请求中更改。此可变参数可以在路径中定义的变量中接受，并传递给绑定到操作装饰器的函数中定义的形式参数。

示例

添加另一个路径装饰器，在路由中使用可变参数，并将hello()函数绑定到 name 参数。根据以下内容修改 main.py。

import uvicorn
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def index():
   return {"message": "Hello World"}
@app.get("/hello/{name}")
async def hello(name):
   return {"name": name}

启动 Uvicorn 服务器并访问 https://127.0.0.1:8000/hello/Tutorialspoint URL。浏览器显示以下 JSON 响应。

{"name":"Tutorialspoint"}

将可变路径参数更改为其他内容，例如 https://127.0.0.1:8000/hello/Python，以便浏览器显示：

{"name":"Python"}

检查 OpenAPI 文档

现在，如果我们通过输入 URL https://127.0.0.1:8000/docs 检查 OpenAPI 文档，它将显示两条路由及其各自的视图函数。单击 /hello/{name} 按钮下方的“试用”按钮，并将 Tutorialspoint 作为 name 参数描述的值，然后单击“执行”按钮。

然后它将显示Curl命令、请求 URL以及服务器响应的详细信息，包括响应正文和响应标头。

一条路由可以有多个用“/”符号分隔的参数。

from fastapi import FastAPI
app = FastAPI()
@app.get("/hello/{name}/{age}")
async def hello(name,age):
   return {"name": name, "age":age}

在这种情况下，/hello是路由，后面跟着两个用花括号括起来的参数。如果浏览器地址栏中给出的 URL 是 https://127.0.0.1:8000/hello/Ravi/20，则 Ravi 和 20 的数据将分别分配给变量 name 和 age。浏览器显示以下 JSON 响应：

{"name":"Ravi","age":"20"}

带类型的路径参数

您可以对要装饰的函数的参数使用 Python 的类型提示。在这种情况下，将 name 定义为 str，将 age 定义为 int。

@app.get("/hello/{name}/{age}")
async def hello(name:str,age:int):
   return {"name": name, "age":age} 

如果类型不匹配，这将导致浏览器在 JSON 响应中显示 HTTP 错误消息。尝试输入 https://127.0.0.1:8000/hello/20/Ravi 作为 URL。浏览器的响应如下：

{
   "detail": [
      {
         "loc": [
            "path",
            "age"
         ],
      "msg": "value is not a valid integer",
      "type": "type_error.integer"
      }
   ]
}

原因很明显，因为age是整数，不能接受字符串值。这也会反映在 Swagger UI（OpenAPI）文档中。

FastAPI - 查询参数

将请求数据传递到服务器的经典方法是将查询字符串附加到 URL。假设服务器上的 Python 脚本 (hello.py) 作为CGI执行，由＆符号连接的一系列键值对构成查询字符串，通过将问号 (?) 作为分隔符附加到 URL 来附加查询字符串。例如：

https://127.0.0.1/cgi-bin/hello.py?name=Ravi&age=20

URL 的尾部（在 (?) 之后）是查询字符串，然后由服务器端脚本解析以进行进一步处理。

如前所述，查询字符串是由 & 符号连接的一系列 parameter=value 对。FastAPI 自动将端点中不是路径参数的部分视为查询字符串，并将其解析成参数及其值。这些参数传递给操作装饰器下面的函数。

示例

from fastapi import FastAPI
app = FastAPI()
@app.get("/hello")
async def hello(name:str,age:int):
   return {"name": name, "age":age}

启动 Uvicorn 服务器并在浏览器中输入此 URL：

https://127.0.0.1:8000/hello?name=Ravi&age=20

您应该获得相同的 JSON 响应。但是，检查表明 FastAPI 检测到 /hello 端点没有路径参数，但有查询参数。

单击“试用”按钮，输入“Ravi”和“20”作为值，然后按“执行”按钮。文档页面现在显示 Curl 命令、请求 URL 以及 HTTP 响应的主体和标头。

示例

您可以对要装饰的函数的参数使用 Python 的类型提示。在这种情况下，将 name 定义为 str，将 age 定义为 int。

from fastapi import FastAPI
app = FastAPI()
@app.get("/hello/{name}")
async def hello(name:str,age:int):
   return {"name": name, "age":age}

尝试输入 https://127.0.0.1:8000/docs 作为 URL。这将打开 Swagger UI（OpenAPI）文档。参数“name”是路径参数，“age”是查询参数

FastAPI - 参数验证

可以对 URL 的路径参数和查询参数应用验证条件。为了对路径参数应用验证条件，您需要导入 Path 类。除了参数的默认值外，您还可以为字符串参数指定最大长度和最小长度。

from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/hello/{name}")
async def hello(name:str=Path(...,min_length=3,
max_length=10)):
   return {"name": name}

如果浏览器 URL 包含长度小于 3 或大于 10 的参数，如 (https://127.0.0.1:8000/hello/Tutorialspoint)，则会出现相应的错误消息，例如：

{
   "detail": [
      {
         "loc": [
            "path",
            "name"
         ],
         "msg": "ensure this value has at most 10 characters",
         "type": "value_error.any_str.max_length",
         "ctx": {
            "limit_value": 10
         }
      }
   ]
}

OpenAPI 文档也显示了应用的验证：

也可以对数字参数应用验证规则，使用以下运算符：

• gt - 大于

• ge - 大于或等于

• lt - 小于

• le - 小于或等于

让我们修改上面的操作装饰器，将 age 作为路径参数包含在内，并应用验证。

from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/hello/{name}/{age}")
async def hello(*, name: str=Path(...,min_length=3 , max_length=10), age: int = Path(..., ge=1, le=100)):
   return {"name": name, "age":age}

在这种情况下，对参数 name 和 age 都应用了验证规则。如果输入的 URL 是 https://127.0.0.1:8000/hello/hi/110，则 JSON 响应显示以下验证失败说明：

{
   "detail": [
      {
         "loc": [
            "path",
            "name"
         ],
         "msg": "ensure this value has at least 3 characters",
         "type": "value_error.any_str.min_length",
         "ctx": {
            "limit_value": 3
         }
      },
      {
         "loc": [
            "path",
            "age"
         ],
         "msg": "ensure this value is less than or equal to 100",
         "type": "value_error.number.not_le",
         "ctx": {
            "limit_value": 100
         }
      }
   ]
}

Swagger UI 文档也识别了约束。

查询参数也可以应用验证规则。您需要将它们指定为 Query 类构造函数参数的一部分。

让我们在上面的函数中添加一个名为percent的查询参数，并应用验证规则 ge=0（即大于或等于 0）和lt=100（小于或等于 100）

from fastapi import FastAPI, Path, Query
@app.get("/hello/{name}/{age}")
async def hello(*, name: str=Path(...,min_length=3 ,
max_length=10), \
      age: int = Path(..., ge=1, le=100), \
      percent:float=Query(..., ge=0, le=100)):
   return {"name": name, "age":age}

如果输入的 URL 是 https://127.0.0.1:8000/hello/Ravi/20?percent=79，则浏览器显示以下 JSON 响应：

{"name":"Ravi","age":20}

FastAPI 正确地将 percent 识别为应用了验证条件的查询参数。它在 OpenAPI 文档中反映如下：

虽然客户端可以使用 GET 方法将路径和查询参数发送到 API 服务器，但我们需要应用 POST 方法才能将一些二进制数据作为 HTTP 请求的一部分发送。此二进制数据可以是任何 Python 类的对象形式。它构成请求正文。FastAPI 使用 Pydantic 库来实现此目的。

FastAPI - Pydantic

Pydantic是一个用于数据解析和验证的 Python 库。它使用 Python 新版本（3.6 及更高版本）的类型提示机制，并在运行时验证类型。Pydantic 定义了BaseModel类。它充当创建用户定义模型的基类。

以下代码将 Student 类定义为基于 BaseModel 的模型。

from typing import List
from pydantic import BaseModel
class Student(BaseModel):
   id: int
   name :str
   subjects: List[str] = []

Student类的属性用类型提示声明。请注意，subjects 属性是 typing 模块中定义的 List 类型，并且是内置列表类型。

我们可以使用具有匹配结构的字典填充 Student 类对象，如下所示：

>>> data = {
   'id': 1,
   'name': 'Ravikumar',
   'subjects': ["Eng", "Maths", "Sci"],
}
>>> s1=Student(**data)
>>> print (s1)
id=1 name='Ravikumar' subjects=['Eng', 'Maths', 'Sci']
>>> s1
Student(id=1, name='Ravikumar', subjects=['Eng', 'Maths', 'Sci'])
>>> s1.dict()
{'id': 1, 'name': 'Ravikumar', 'subjects': ['Eng', 'Maths', 'Sci']}

Pydantic将在可能的情况下自动转换数据类型。例如，即使字典中的 id 键被分配了一个数字的字符串表示形式（例如“123”），它也会将其强制转换为整数。但是，在不可能的情况下，将引发异常。

>>> data = {
   'id': [1,2],
   'name': 'Ravikumar',
   'subjects': ["Eng", "Maths", "Sci"],
}
>>> s1=Student(**data)
Traceback (most recent call last):
   File "<pyshell#13>", line 1, in <module>
      s1=Student(**data)
   File "pydantic\main.py", line 406, in
pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error
for Student
id
   value is not a valid integer (type=type_error.integer)

Pydantic 还包含一个 Field 类，用于声明模型属性的元数据和验证规则。首先修改 Student 类，以便如下在“name”属性上应用 Field 类型：

from typing import List
from pydantic import BaseModel, Field
class Student(BaseModel):
   id: int
   name :str = Field(None, title="The description of the item", max_length=10)
   subjects: List[str] = []

按如下所示填充数据。这里的 name 超过了规定的max_length。Pydantic 按预期抛出ValidationError。

>>> data = {
   'id': 1,
   'name': 'Ravikumar Sharma',
   'subjects': ["Eng", "Maths", "Sci"],
}
>>> s1=Student(**data)
Traceback (most recent call last):
   File "<pyshell#28>", line 1, in <module>
      s1=Student(**data)
   File "pydantic\main.py", line 406, in
pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for Student
name
   ensure this value has at most 10 characters
   (type=value_error.any_str.max_length; limit_value=10)

Pydantic 模型可以与SQLAlchemy或Peewee等 ORM 模型映射。

FastAPI - 请求体

我们现在将使用 Pydantic 模型对象作为客户端请求的请求正文。如前所述，我们需要为此使用 POST 操作装饰器。

import uvicorn
from fastapi import FastAPI
from typing import List
from pydantic import BaseModel, Field
app = FastAPI()
class Student(BaseModel):
   id: int
   name :str = Field(None, title="name of student", max_length=10)
   subjects: List[str] = []
@app.post("/students/")
async def student_data(s1: Student):
   return s1

可以看到，student_data() 函数被 @app.post() 装饰器装饰，该装饰器的 URL 端点为 "/students/"。它从客户端的请求中接收 Student 类的对象作为 Body 参数。要测试此路由，请启动 Uvicorn 服务器并在浏览器中访问 https://127.0.0.1:8000/docs 打开 Swagger UI 文档。

文档标识 "/students/" 路由与 student_data() 函数和 POST 方法相关联。在 schemas 部分，将列出 Student 模型。

展开其前面的节点以显示模型的结构。

单击 Try it out 按钮以在请求正文中填写测试值。

单击 Execute 按钮并获取服务器的响应值。

虽然 Pydantic 模型会自动填充请求正文，但也可以使用单个值向其添加属性。为此，我们需要将 Body 类对象用作要装饰的操作函数的参数。

首先，我们需要从 fastapi 中导入 Body 类。如下例所示，在 @app.post() 装饰器下方的 student_data() 函数定义中声明 'name' 和 'marks' 为 Body 参数。

import uvicorn
from fastapi import FastAPI, Body
@app.post("/students")
async def student_data(name:str=Body(...),
marks:int=Body(...)):
   return {"name":name,"marks": marks}

如果我们检查 Swagger UI 文档，我们应该能够找到此 POST 方法与 student_data() 函数相关联，并且具有两个参数的请求正文。

还可以声明一个操作函数，使其具有路径和/或查询参数以及请求正文。让我们修改 student_data() 函数，使其具有路径参数 'college'、查询参数 'age' 和正文参数 Student 模型对象。

@app.post("/students/{college}")
async def student_data(college:str, age:int, student:Student):
   retval={"college":college, "age":age, **student.dict()}
   return retval

该函数将 college 和 age 参数的值与 Student 对象的字典表示形式一起添加，并将其作为响应返回。我们可以如下检查 API 文档 -

可以看到，college 是路径参数，age 是查询参数，Student 模型是请求正文。

FastAPI - 模板

默认情况下，FastAPI 向客户端呈现 JSON 响应。但是，它可以转换为 HTML 响应。为此，FastAPI 在 fastapi.responses 模块中定义了 HTMLResponse 类。我们需要将 response_class 作为附加参数添加到操作装饰器中，并将 HTMLResponse 对象作为其值。

在下面的示例中，@app.get() 装饰器具有 "/hello/" 端点，并且 HTMLResponse 作为 response_class。在 hello() 函数内部，我们有一个 Hello World 消息的 HTML 代码的字符串表示形式。该字符串以 HTML 响应的形式返回。

from fastapi.responses import HTMLResponse
from fastapi import FastAPI
app = FastAPI()
@app.get("/hello/")
async def hello():
   ret='''
<html>
<body>
<h2>Hello World!</h2>
</body>
</html>
'''
   return HTMLResponse(content=ret)

检查 API 文档后，可以发现服务器的响应正文为 HTML。

请求 URL (https://127.0.0.1:8000/hello/) 也应该在浏览器中呈现消息。但是，呈现原始 HTML 响应非常繁琐。或者，可以呈现预构建的 HTML 页面作为模板。为此，我们需要使用 Web 模板库。

Web 模板库具有一个模板引擎，该引擎合并一个具有占位符变量的静态网页。来自任何来源（例如数据库）的数据被合并以动态生成和呈现网页。FastAPI 没有任何预打包的模板库。因此，您可以自由使用任何适合您需求的库。在本教程中，我们将使用 jinja2，这是一个非常流行的 Web 模板库。让我们首先使用 pip 安装程序安装它。

pip3 install jinja2

FastAPI 对 Jinja 模板的支持以 jinja2Templates 类的形式出现，该类在 fastapi.templates 模块中定义。

from fastapi.templating import Jinja2Templates

要声明一个模板对象，应将存储 html 模板的文件夹作为参数提供。在当前工作目录中，我们将创建一个 'templates' 目录。

templates = Jinja2Templates(directory="templates")

一个简单的网页 'hello.html' 用于呈现 Hello World 消息，也放在 'templates' 文件夹中。

<html>
<body>
<h2>Hello World!</h2>
</body>
</html>

我们现在将从此页面呈现 html 代码作为 HTMLResponse。让我们修改 hello() 函数如下 -

from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi import FastAPI, Request
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@app.get("/hello/", response_class=HTMLResponse)
async def hello(request: Request):
   return templates.TemplateResponse("hello.html", {"request": request})

这里，模板对象的 templateResponse() 方法收集模板代码和请求上下文以呈现 http 响应。当我们启动服务器并访问 https://127.0.0.1:8000/hello/ URL 时，我们在浏览器中看到了 Hello World 消息，这实际上是 hello.html 的输出。

如前所述，jinja2 模板允许在 HTML 代码中嵌入某些占位符。jinja2 代码元素放在花括号内。一旦浏览器的 HTML 解析器遇到此代码，模板引擎就会接管并使用 HTTP 响应提供的变量数据填充这些代码元素。Jinja2 提供以下代码元素 -

• {% %} – 语句

• {{ }} – 打印到模板输出的表达式

• {# #} − 注释，不包含在模板输出中

• # # # − 行语句

hello.html 如下修改，通过替换 name 参数来显示动态消息。

<html>
<body>
<h2>Hello {{name}} Welcome to FastAPI</h2>
</body>
</html>

操作函数 hello() 也被修改为接受 name 作为路径参数。TemplateResponse 还应包含 “name”:name 的 JSON 表示形式以及请求上下文。

from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi import FastAPI, Request
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@app.get("/hello/{name}", response_class=HTMLResponse)
async def hello(request: Request, name:str):
   return templates.TemplateResponse("hello.html", {"request": request, "name":name})

重新启动服务器并转到 https://127.0.0.1:8000/hello/Kiran。浏览器现在使用此 URL 中的路径参数填充 jinja2 占位符。

FastAPI - 静态文件

通常需要在模板响应中包含一些资源，即使存在某些动态数据，这些资源也保持不变。此类资源称为静态资源。媒体文件（.png、.jpg 等）、用于执行某些前端代码的 JavaScript 文件或用于格式化 HTML（.CSS 文件）的样式表是静态文件的示例。

为了处理静态文件，您需要一个名为 aiofiles 的库。

pip3 install aiofiles

接下来，从 fastapi.staticfiles 模块导入 StaticFiles 类。其对象是 FastAPI 应用程序对象的 mount() 方法的参数之一，用于将当前应用程序文件夹中的 "static" 子文件夹分配给存储和提供应用程序的所有静态资源。

app.mount(app.mount("/static", StaticFiles(directory="static"), name="static")

示例

在下面的示例中，FastAPI 徽标将在 hello.html 模板中呈现。因此，首先将“fa-logo.png”文件放置在 static 文件夹中。现在可以将其用作 HTML 代码中 <img> 标记的 src 属性。

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
app = FastAPI()
templates = Jinja2Templates(directory="templates")
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/hello/{name}", response_class=HTMLResponse)
async def hello(request: Request, name:str):
   return templates.TemplateResponse("hello.html", {"request": request, "name":name})

\templates\hello.html 的 HTML 代码如下 -

<html>
   <body>
      <h2>Hello {{name}} Welcome to FastAPI</h2>
      <img src="{{ url_for('static', path='fa-logo.png') }}" alt="" width="300">
   </body>
</html>
</pre>

运行 Uvicorn 服务器并访问 URL https://127.0.0.1/hello/Vijay。徽标将显示在浏览器窗口中，如所示。

示例

这是另一个静态文件的示例。JavaScript 代码 hello.js 包含 myfunction() 的定义，该定义将在以下 HTML 脚本（\templates\hello.html）的 onload 事件上执行。

<html>
   <head>
      <title>My Website</title>
      <script src="{{ url_for('static', path='hello.js') }}"></script>
   </head>
   <body onload="myFunction()">
      <div id="time" style="text-align:right; width="100%"></div>
      <h1><div id="ttl">{{ name }}</div></h1>
   </body>
</html>

hello.js 代码如下 - (\static\hello.js)

function myFunction() {
   var today = new Date();
   var h = today.getHours();
   var m = today.getMinutes();
   var s = today.getSeconds();
   var msg="";
   if (h<12) {
      msg="Good Morning, ";
   }
   if (h>=12 && h<18) {
      msg="Good Afternoon, ";
   }
   if (h>=18) {
      msg="Good Evening, ";
   }
   var x=document.getElementById('ttl').innerHTML;
   document.getElementById('ttl').innerHTML = msg+x;
   document.getElementById('time').innerHTML = h + ":" + m + ":" + s;
}

该函数检测当前时间的值，并根据一天中的时间为 msg 变量分配相应的值（早上好、下午好或晚上好）。

保存 /static/hello.js，修改 \templates\hello.html 并重新启动服务器。浏览器应显示当前时间及其下方的相应消息。

FastAPI - HTML 表单模板

让我们向我们的应用程序添加另一个路由 "/login"，该路由呈现一个具有简单登录表单的 html 模板。登录页面的 HTML 代码如下 -

<html>
   <body>
      <form action="/submit" method="POST">
         <h3>Enter User name</h3>
         <p><input type='text' name='nm'/></p>
         <h3>Enter Password</h3>
         <p><input type='password' name='pwd'/></p>
         <p><input type='submit' value='Login'/></p>
      </form>
   </body>
</html>

请注意，action 参数设置为 "/submit" 路由，action 设置为 POST。这对于进一步讨论非常重要。

在 main.py 文件中添加 login() 函数，如下所示 -

@app.get("/login/", response_class=HTMLResponse)
async def login(request: Request):
   return templates.TemplateResponse("login.html", {"request": request})

URL https://127.0.0.1:8000/login 将呈现登录表单，如下所示 -

FastAPI - 访问表单数据

现在我们将了解如何在 FastAPI 操作函数中访问 HTML 表单数据。在上面的示例中，/login 路由呈现登录表单。用户输入的数据通过 POST 作为请求方法提交到 /submit URL。现在我们必须提供一个视图函数来处理用户提交的数据。

FastAPI 具有一个 Form 类来处理通过提交 HTML 表单作为请求接收的数据。但是，您需要安装 python-multipart 模块。它是 Python 的流式多部分表单解析器。

pip3 install python-multipart

将 Form 类添加到从 FastAPI 导入的资源中。

from fastapi import Form

让我们定义一个由 @app.post() 装饰的 submit() 函数。为了接收表单数据，声明两个 Form 类型的参数，其名称与表单属性相同。

@app.post("/submit/")
async def submit(nm: str = Form(...), pwd: str = Form(...)):
   return {"username": nm}

填充文本字段后，按提交。浏览器将重定向到 /submit URL 并呈现 JSON 响应。检查 /submit 路由的 Swagger API 文档。它正确地将 nm 和 pwd 识别为请求正文参数，并将表单的“媒体类型”识别为 application/x-www-form-urlencoded。

甚至可以使用 HTML 表单数据填充和返回 Pydantic 模型。在以下代码中，我们声明 User 类作为 Pydantic 模型，并将其对象作为服务器的响应发送。

from pydantic import BaseModel
class User(BaseModel):
   username:str
   password:str
@app.post("/submit/", response_model=User)
async def submit(nm: str = Form(...), pwd: str = Form(...)):
   return User(username=nm, password=pwd)

FastAPI - 上传文件

首先，要将文件发送到服务器，您需要将 HTML 表单的 enctype 用作 multipart/form-data，并将输入类型用作文件以呈现按钮，单击该按钮允许您从文件系统中选择文件。

<html>
   <body>
      <form action="https://127.0.0.1:8000/uploader" method="POST" enctype="multipart/form-data">
         <input type="file" name="file" />
         <input type="submit"/>
      </form>
   </body>
</html>

请注意，表单的 action 参数设置为端点 https://127.0.0.1:8000/uploader，并且方法设置为 POST。

此 HTML 表单将使用以下代码作为模板呈现 -

from fastapi import FastAPI, File, UploadFile, Request
import uvicorn
import shutil
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@app.get("/upload/", response_class=HTMLResponse)
async def upload(request: Request):
   return templates.TemplateResponse("uploadfile.html", {"request": request})

访问 https://127.0.0.1:8000/upload/。您应该会看到带有 选择文件 按钮的表单。单击它以打开要上传的文件。

上传操作由 FastAPI 中的 UploadFile 函数处理。

from fastapi import FastAPI, File, UploadFile
import shutil
@app.post("/uploader/")
async def create_upload_file(file: UploadFile = File(...)):
   with open("destination.png", "wb") as buffer:
      shutil.copyfileobj(file.file, buffer)
   return {"filename": file.filename}

我们将使用 Python 中的 shutil 库以 destination.png 的名称将接收到的文件复制到服务器位置。

FastAPI - Cookie 参数

cookie 是 HTTP 标头之一。Web 服务器向客户端发送响应，除了请求的数据外，它还会插入一个或多个 cookie。cookie 是少量数据，存储在客户端的机器上。在来自同一客户端的后续连接请求中，此 cookie 数据也会与 HTTP 请求一起附加。

cookie 用于记录有关客户端浏览的信息。在 HTTP 协议的无状态通信中，cookie 是检索有状态信息的可靠方法。

在 FastAPI 中，cookie 参数使用 set_cookie() 方法设置在响应对象上。

response.set_cookie(key, value)

示例

这是一个 set_cookie() 方法的示例。我们有一个名为 content 的 JSON 响应对象。在其上调用 set_cookie() 方法以将 cookie 设置为 key="usrname" 和 value="admin" -

from fastapi import FastAPI
from fastapi.responses import JSONResponse
app = FastAPI()
@app.post("/cookie/")
def create_cookie():
   content = {"message": "cookie set"}
   response = JSONResponse(content=content)
   response.set_cookie(key="username", value="admin")
   return response

要在后续访问中读回 cookie，请在 FastAPI 库中使用 Cookie 对象。

from fastapi import FastAPI, Cookie
app = FastAPI()
@app.get("/readcookie/")
async def read_cookie(username: str = Cookie(None)):
   return {"username": username}

在 Swagger API 中检查这两个端点。这两个路由是 "/cookies" 和 "/readcookie"。执行绑定到 "/cookies" 的 create_cookie() 函数。响应只是内容，尽管 cookie 已设置。

当执行read_cookie()函数时，cookie会被读取并作为响应返回。此外，请注意文档将用户名标识为cookie参数。

FastAPI - 头部参数

为了读取客户端请求中HTTP头部的值，需要从FastAPI库导入Header对象，并在操作函数定义中声明一个Header类型的参数。参数名称应与转换为camel_case的HTTP头部名称匹配。

在下面的示例中，需要检索“accept-language”头部。由于Python不允许在标识符名称中使用“-”（短横线），因此将其替换为“_”（下划线）

from typing import Optional
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/headers/")
async def read_header(accept_language: Optional[str] = Header(None)):
   return {"Accept-Language": accept_language} 

如下Swagger文档所示，检索到的头部显示在响应体中。

您可以在响应对象中推送自定义和预定义的头部。操作函数应具有一个Response类型的参数。为了设置自定义头部，其名称应以“X”为前缀。在以下情况下，添加了一个名为“X-Web-Framework”的自定义头部和一个预定义的头部“Content-Language”，以及操作函数的响应。

from fastapi import FastAPI
from fastapi.responses import JSONResponse
app = FastAPI()
@app.get("/rspheader/")
def set_rsp_headers():
   content = {"message": "Hello World"}
   headers = {"X-Web-Framework": "FastAPI", "Content-Language": "en-US"}
   return JSONResponse(content=content, headers=headers)

新添加的头部将出现在文档的响应头部部分。

FastAPI - 响应模型

操作函数向客户端返回一个JSON响应。响应可以是Python基本类型，例如数字、字符串、列表或字典等。它也可以是Pydantic模型的形式。为了使函数返回模型对象，操作装饰器应声明一个respone_model参数。

借助response_model，FastAPI将输出数据转换为模型类的结构。它验证数据，并在OpenAPI路径操作中为响应添加JSON Schema。

response_model参数的重要优势之一是，我们可以通过从模型中选择字段来格式化输出，并将响应转换为输出模型。

示例

在下面的示例中，POST操作装饰器以student类（BaseModel的子类）对象的格式接收请求体。由于此类中的一个字段，即marks（分数列表）在响应中不需要，因此我们定义了另一个名为percent的模型，并将其用作response_model参数。

from typing import List
from fastapi import FastAPI
from pydantic import BaseModel, Field
app = FastAPI()
class student(BaseModel):
   id: int
   name :str = Field(None, title="name of student", max_length=10)
   marks: List[int] = []
   percent_marks: float
class percent(BaseModel):
   id:int
   name :str = Field(None, title="name of student", max_length=10)
   percent_marks: float
@app.post("/marks", response_model=percent)
async def get_percent(s1:student):
   s1.percent_marks=sum(s1.marks)/2
   return s1

如果我们检查Swagger文档，它显示“/marks”路由获取student类对象作为请求体。使用适当的值填充属性并执行get_percent()函数。

服务器响应被转换为percent类，因为它已被用作response_model。

FastAPI - 嵌套模型

Pydantic模型的每个属性都有一个类型。类型可以是内置的Python类型或模型本身。因此，可以使用特定的属性名称、类型和验证来声明嵌套的JSON“对象”。

示例

在下面的示例中，我们构造了一个customer模型，其中一个属性为product模型类。product模型又具有supplier类的属性。

from typing import Tuple
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class supplier(BaseModel):
   supplierID:int
   supplierName:str
class product(BaseModel):
   productID:int
   prodname:str
   price:int
   supp:supplier
class customer(BaseModel):
   custID:int
   custname:str
   prod:Tuple[product]

下面的POST操作装饰器将customer模型的对象呈现为服务器响应。

@app.post('/invoice')
async def getInvoice(c1:customer):
   return c1

Swagger UI页面显示了三个架构，分别对应于三个BaseModel类。

展开Customer架构以显示所有节点，如下所示：

"/invoice"路由的示例响应如下：

{
   "custID": 1,
   "custname": "Jay",
   "prod": [
      {
         "productID": 1,
         "prodname": "LAPTOP",
         "price": 40000,
         "supp": {
            "supplierID": 1,
            "supplierName": "Dell"
         }
      }
   ]
}

FastAPI - 依赖项

FastAPI的内置依赖注入系统使得在构建API时更容易集成组件。在编程中，依赖注入是指一个对象接收其依赖的其他对象。其他对象称为依赖项。依赖注入具有以下优点：

• 重用相同的共享逻辑

• 共享数据库连接

• 执行身份验证和安全功能

假设一个FastAPI应用程序有两个操作函数，它们都具有相同的查询参数id、name和age。

from fastapi import FastAPI
app = FastAPI()
@app.get("/user/")
async def user(id: str, name: str, age: int):
   return {"id": id, "name": name, "age": age}
@app.get("/admin/")
async def admin(id: str, name: str, age: int):
   return {"id": id, "name": name, "age": age}

如果发生任何更改，例如添加/删除查询参数，则需要更改两个路由装饰器和函数。

FastAPI提供Depends类，其对象在这种情况下用作公共参数。首先从FastAPI导入Depends并定义一个函数来接收这些参数：

async def dependency(id: str, name: str, age: int):
   return {"id": id, "name": name, "age": age}

现在，我们可以将此函数的返回值用作操作函数中的参数

@app.get("/user/")
async def user(dep: dict = Depends(dependency)):
   return dep

对于每个新请求，FastAPI都会使用相应的参数调用依赖函数，返回结果，并将结果分配给您的操作。

您可以使用类来管理依赖项，而不是函数。声明一个类，其中id、name和age作为属性。

class dependency:
   def __init__(self, id: str, name: str, age: int):
      self.id = id
      self.name = name
      self.age = age 

使用此类作为参数的类型。

@app.get("/user/")
async def user(dep: dependency = Depends(dependency)):
   return dep
@app.get("/admin/")
async def admin(dep: dependency = Depends(dependency)):
   return dep 

在这里，我们在操作函数中使用了依赖注入。它也可以用作操作装饰器。例如，我们想要检查查询参数age的值是否小于21。如果是，则应抛出异常。因此，我们编写一个函数来检查它并将其用作依赖项。

async def validate(dep: dependency = Depends(dependency)):
   if dep.age > 18:
      raise HTTPException(status_code=400, detail="You are not eligible")
@app.get("/user/", dependencies=[Depends(validate)])
async def user():
   return {"message": "You are eligible"}

在FastAPI依赖管理中，您可以使用yield代替return来添加一些额外的步骤。例如，以下函数使用带yield的数据库依赖项。

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

FastAPI - CORS

跨源资源共享（CORS）是指当在一个客户端浏览器上运行的前端应用程序尝试通过JavaScript代码与后端通信，而后端与前端位于不同的“源”时的情况。此处的源是协议、域名和端口号的组合。因此，https://127.0.0.1和https://127.0.0.1具有不同的源。

如果具有一个源的URL的浏览器发送请求以执行来自另一个源的JavaScript代码，则浏览器会发送一个OPTIONS HTTP请求。

如果后端通过发送适当的头部授权来自此不同源的通信，它将允许前端的JavaScript将其请求发送到后端。为此，后端必须有一个“允许的源”列表。

要显式指定允许的源，请导入CORSMiddleware并将源列表添加到应用程序的中件件中。

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

FastAPI - CRUD操作

REST架构使用HTTP动词或方法对资源进行操作。POST、GET、PUT和DELETE方法分别执行CREATE、READ、UPDATE和DELETE操作。

在下面的示例中，我们将使用Python列表作为内存数据库并在其上执行CRUD操作。首先，让我们设置一个FastAPI应用程序对象并声明一个名为Book的Pydantic模型。

from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
data = []
class Book(BaseModel):
   id: int
   title: str
   author: str
   publisher: str

使用@app.post()装饰器填充此模型的对象，并将其追加到书籍列表中（为书籍列表声明数据）

@app.post("/book")
def add_book(book: Book):
   data.append(book.dict())
   return data

在Swagger UI中，执行此操作函数几次并添加一些数据。

服务器的JSON响应显示了迄今为止添加的书籍列表。

要检索列表，请定义一个绑定到@app.get()装饰器的操作函数，如下所示：

@app.get("/list")
def get_books():
   return data

要检索其id作为路径参数的书籍，请定义get()操作装饰器和get_book()函数，如下所示：

@app.get("/book/{id}")
def get_book(id: int):
   id = id - 1
   return data[id]

/list路由检索所有书籍。

另一方面，在“/book/1”路由中使用“id”作为路径参数。

将检索到“id=1”的书籍，如Swagger UI的服务器响应中所示

接下来，定义@app.put()装饰器，该装饰器修改数据列表中的对象。此装饰器也为id字段具有路径参数。

@app.put("/book/{id}")
def add_book(id: int, book: Book):
   data[id-1] = book
   return data

在swagger UI中检查此操作函数。给出id=1，并在请求体中将publisher的值更改为BPB。

执行后，响应显示更新了新值的id=1的对象的列表。

最后，我们定义@app.delete()装饰器以删除与路径参数对应的对象。

@app.delete("/book/{id}")
def delete_book(id: int):
   data.pop(id-1)
   return data

将id=1作为路径参数并执行函数。

执行后，列表现在仅显示两个对象

FastAPI - SQL 数据库

在上一章中，Python列表已被用作内存数据库，以使用FastAPI执行CRUD操作。相反，我们可以使用任何关系数据库（例如MySQL、Oracle等）来执行存储、检索、更新和删除操作。

我们不使用符合DB-API的数据库驱动程序，而是使用SQLAlchemy作为Python代码和数据库之间的接口（我们将使用SQLite数据库，因为Python对它有内置支持）。SQLAlchemy是一个流行的SQL工具包和对象关系映射器。

对象关系映射是一种编程技术，用于在面向对象编程语言中转换不兼容类型系统之间的数据。通常，面向对象语言（如Python）中使用的类型系统包含非标量类型。但是，大多数数据库产品（如Oracle、MySQL等）中的数据类型都是原始类型，例如整数和字符串。

在ORM系统中，每个类都映射到底层数据库中的一个表。ORM可以为您处理这些问题，而您可以专注于系统逻辑的编程，而不是自己编写乏味的数据库接口代码。

为了使用SQLAlchemy，我们需要首先使用PIP安装程序安装库。

pip install sqlalchemy

SQLAlchemy旨在与为特定数据库构建的DBAPI实现一起使用。它使用方言系统与各种类型的DBAPI实现和数据库进行通信。所有方言都需要安装相应的DBAPI驱动程序。

以下是被包含的方言：

• Firebird

• Microsoft SQL Server

• MySQL

• Oracle

• PostgreSQL

• SQLite

• Sybase

由于我们将使用SQLite数据库，因此我们需要为名为test.db的数据库创建一个数据库引擎。从sqlalchemy模块导入create_engine()函数。

from sqlalchemy import create_engine
from sqlalchemy.dialects.sqlite import *
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args = {"check_same_thread": False})

为了与数据库交互，我们需要获取其句柄。会话对象是数据库的句柄。Session类使用sessionmaker()定义，这是一个可配置的会话工厂方法，它绑定到引擎对象。

from sqlalchemy.orm import sessionmaker, Session
session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

接下来，我们需要一个声明性基类，它在声明性系统中存储类的目录和映射表。

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

Books（Base的子类）映射到数据库中的book表。Books类中的属性对应于目标表中列的数据类型。请注意，id属性对应于book表中的主键。

from sqlalchemy import Column, Integer, String
class Books(Base):
   __tablename__ = 'book'
   id = Column(Integer, primary_key=True, nullable=False)
   title = Column(String(50), unique=True)
   author = Column(String(50))
   publisher = Column(String(50))
   Base.metadata.create_all(bind=engine)

create_all()方法在数据库中创建相应的表。

现在，我们必须声明一个与声明性基子类（上面定义的Books类）对应的Pydantic模型。

from typing import List
from pydantic import BaseModel, constr
class Book(BaseModel):
   id: int
   title: str
   author:str
   publisher: str
   class Config:
      orm_mode = True

请注意，在config类中使用orm_mode=True表示它已映射到SQLAlchemy的ORM类。

其余代码与内存CRUD操作类似，区别在于操作函数通过SQLalchemy接口与数据库交互。下面定义了FastAPI应用程序对象上的POST操作：

from fastapi import FastAPI, Depends
app=FastAPI()
def get_db():
   db = session()
   try:
      yield db
   finally:
   db.close()
@app.post('/add_new', response_model=Book)
def add_book(b1: Book, db: Session = Depends(get_db)):
   bk=Books(id=b1.id, title=b1.title, author=b1.author,
publisher=b1.publisher)
   db.add(bk)
   db.commit()
   db.refresh(bk)
   return Books(**b1.dict())

首先建立数据库会话。来自POST请求体的数据作为新行添加到book表中。执行add_book()操作函数以将示例数据添加到books表中。要进行验证，您可以使用SQLiteStudio，这是一个用于SQLite数据库的GUI工具。

定义了两个用于GET操作的操作函数，一个用于获取所有记录，另一个用于匹配路径参数的记录。

以下是绑定到/list路由的get_books()函数。执行后，其服务器响应为所有记录的列表。

@app.get('/list', response_model=List[Book])
def get_books(db: Session = Depends(get_db)):
   recs = db.query(Books).all()
   return recs

/book/{id}路由调用get_book()函数，其中id作为路径参数。SQLAlchemy的查询返回与给定id对应的对象。

@app.get('/book/{id}', response_model=Book)
def get_book(id:int, db: Session = Depends(get_db)):
   return db.query(Books).filter(Books.id == id).first()

下图显示了从Swagger UI执行get_books()函数的结果。

更新和删除操作由update_book()函数（当访问/update/{id}路由时执行）和del_book()函数（当在URL中给出路由/delete/{id}时调用）执行。

@app.put('/update/{id}', response_model=Book)
def update_book(id:int, book:Book, db: Session = Depends(get_db)):
   b1 = db.query(Books).filter(Books.id == id).first()
   b1.id=book.id
   b1.title=book.title
   b1.author=book.author
   b1.publisher=book.publisher
   db.commit()
   return db.query(Books).filter(Books.id == id).first()
@app.delete('/delete/{id}')
def del_book(id:int, db: Session = Depends(get_db)):
   try:
      db.query(Books).filter(Books.id == id).delete()
      db.commit()
   except Exception as e:
      raise Exception(e)
   return {"delete status": "success"}

如果您打算使用任何其他数据库来代替SQLite，则只需要相应地更改方言定义。例如，要使用MySQL数据库和pymysql驱动程序，请将引擎对象的语句更改为以下内容：

engine = create_engine('mysql+pymysql://user:password@localhost/test')

FastAPI - 使用 MongoDB

FastAPI还可以使用NoSQL数据库（如MongoDB、Cassandra、CouchDB等）作为REST应用程序CRUD操作的后端。在本主题中，我们将了解如何在FastAPI应用程序中使用MongoDB。

MongoDB 是一个面向文档的数据库，其中半结构化的文档以 JSON 等格式存储。文档可以包含许多不同的键值对、键数组对，甚至嵌套文档。它类似于 Python 字典对象，是一组键值对。一个或多个这样的文档存储在一个集合中。

MongoDB 中的集合相当于关系数据库中的表。但是，MongoDB（以及所有 NoSQL 数据库）没有预定义的模式。文档类似于基于 SQL 的关系数据库中表中的一行。每个文档可能具有可变数量的键值对。因此，MongoDB 是一个无模式数据库。

要在 FastAPI 中使用 MongoDB，必须在机器上安装 MongoDB 服务器。我们还需要安装 **PyMongo**，这是 MongoDB 的官方 Python 驱动程序。

pip3 install pymongo

在通过 Python 和 FastAPI 代码与 MongoDB 数据库交互之前，请确保 MongoDB 正在运行，方法是发出以下命令（假设 MongoDB 服务器安装在 e:\mongodb 文件夹中）。

E:\mongodb\bin>mongod
..
waiting for connections on port 27017

PyMongo 模块中的 **MongoClient** 类的对象是 Python 用于与 MongoDB 服务器交互的句柄。

from pymongo import MongoClient
client=MongoClient()

我们将 Book 定义为 BaseModel 类以填充请求体（与 SQLite 示例中使用的相同）

from pydantic import BaseModel
from typing import List
class Book(BaseModel):
   bookID: int
   title: str
   author:str
   publisher: str

设置 FastAPI 应用程序对象 -

from fastapi import FastAPI, status
app = FastAPI()

POST 操作装饰器将 **"/add_new"** 作为 URL 路由，并执行 **add_book()** 函数。它将 Book BaseModel 对象解析为字典，并在 test 数据库的 BOOK_COLLECTION 中添加一个文档。

@app.post("/add_new", status_code=status.HTTP_201_CREATED)
def add_book(b1: Book):
   """Post a new message to the specified channel."""
   with MongoClient() as client:
      book_collection = client[DB][BOOK_COLLECTION]
      result = book_collection.insert_one(b1.dict())
      ack = result.acknowledged
      return {"insertion": ack}

通过访问 https://127.0.0.1:8000/docs 使用 Swagger UI 的 Web 界面添加一些文档。您可以在 MongoDB 的 Compass GUI 前端验证集合。

要检索所有书籍的列表，让我们包含以下获取操作函数 **- get_books()**。当访问 **"/books"** URL 路由时，将执行此函数。

@app.get("/books", response_model=List[str])
def get_books():
   """Get all books in list form."""
   with MongoClient() as client:
      book_collection = client[DB][BOOK_COLLECTION]
      booklist = book_collection.distinct("title")
      return booklist

在这种情况下，服务器响应将是书籍集合中所有标题的列表。

[
   "Computer Fundamentals",
   "Python Cookbook",
   "Let Us Python"
]

以下 GET 装饰器检索与给定 ID 作为路径参数对应的书籍文档 -

@app.get("/books/{id}", response_model=Book)
def get_book(id: int):
   """Get all messages for the specified channel."""
   with MongoClient() as client:
      book_collection = client[DB][BOOK_COLLECTION]
      b1 = book_collection.find_one({"bookID": id})
      return b1

Swagger UI 文档页面显示以下界面 -

执行上述函数时，服务器的 JSON 响应如下 -

FastAPI - 使用 GraphQL

Facebook 于 2012 年开发了 **GraphQL**，这是一种新的 API 标准，旨在优化 RESTful API 调用。GraphQL 是 API 的数据查询和操作语言。与 REST 相比，GraphQL 更灵活、更高效、更准确。GraphQL 服务器仅提供一个端点，并以客户端所需的确切数据进行响应。

由于 GraphQL 与 ASGI 兼容，因此可以轻松地将其集成到 FastAPI 应用程序中。GraphQL 有许多 Python 库。其中一些列在下面 -

• Strawberry

• Ariadne

• Tartiflette

• Graphene

FastAPI 的官方文档建议使用 Strawberry 库，因为它的设计也基于类型注释（就像 FastAPI 本身一样）。

为了将 GraphQL 集成到 FastAPI 应用程序中，首先将 Python 类装饰为 Strawberry 类型。

@strawberry.type
class Book:
   title: str
   author: str
   price: int

接下来，声明一个包含返回 Book 对象的函数的 **Query** 类。

@strawberry.type
class Query:
   @strawberry.field
   def book(self) -> Book:
   return Book(title="Computer Fundamentals", author="Sinha", price=300)

使用此 Query 类作为参数以获取 **Strawberry.Schema** 对象

schema = strawberry.Schema(query=Query)

然后声明 GraphQL 类和 FastAPI 应用程序类的对象。

graphql_app = GraphQL(schema)
app = FastAPI()

最后，将路由添加到 FastAPI 对象并运行服务器。

app.add_route("/book", graphql_app)
app.add_websocket_route("/book", graphql_app)

在浏览器中访问 https://127.0.0.1:8000/book。将打开一个浏览器内 GraphQL IDE。

在注释部分下方，使用 Graphiql IDE 的 Explorer 栏输入以下查询。运行查询以在输出窗格中显示结果。

FastAPI - Websockets

**WebSocket** 是客户端和服务器之间持久的连接，用于在两者之间提供双向、**全双工**通信。通信通过单个 TCP/IP 套接字连接通过 HTTP 进行。可以将其视为 HTTP 的升级，而不是协议本身。

HTTP 的局限性之一在于它是一种严格的半双工或单向协议。另一方面，使用 WebSockets，我们可以发送基于消息的数据，类似于 UDP，但具有 TCP 的可靠性。WebSocket 使用 HTTP 作为初始传输机制，但在收到 HTTP 响应后保持 TCP 连接活动。相同的连接对象可用于客户端和服务器之间的双向通信。因此，可以使用 WebSocket API 构建实时应用程序。

FastAPI 通过 FastAPI 模块中的 WebSocket 类支持 WebSockets。以下示例演示了 FastAPI 应用程序中 WebSocket 的功能。

首先，我们有一个 **index()** 函数，用于呈现模板（socket.html）。它绑定到 "/" 路由。HTML 文件 socket.html 放在“templates”文件夹中。

main.py

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
templates = Jinja2Templates(directory="templates")
from fastapi.staticfiles import StaticFiles
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
   return templates.TemplateResponse("socket.html", {"request": request})

模板文件呈现一个文本框和一个按钮。

socket.html

<!DOCTYPE html>
<html>
   <head>
      <title>Chat</title>
      <script src="{{ url_for('static', path='ws.js') }}"></script>
   </head>
   <body>
      <h1>WebSocket Chat</h1>
      <form action="" onsubmit="sendMessage(event)">
      <input type="text" id="messageText" autocomplete="off"/>
      <button>Send</button>
      </form>
      <ul id='messages'>
      </ul>
   </body>
</html>

在 socket.html 内部，有一个对 JavaScript 函数的调用，该函数将在表单提交时执行。因此，要服务 JavaScript，首先安装“static”文件夹。JavaScript 文件 ws.js 放在“static”文件夹中。

ws.js

var ws = new WebSocket("ws://127.0.0.1:8000/ws");
ws.onmessage = function(event) {
   var messages = document.getElementById('messages')
   var message = document.createElement('li')
   var content = document.createTextNode(event.data)
   message.appendChild(content)
   messages.appendChild(message)
};
function sendMessage(event) {
   var input = document.getElementById("messageText")
   ws.send(input.value)
   input.value = ''
   event.preventDefault()
}

加载 JavaScript 代码后，它将在“ws://127.0.0.1:8000/ws”处创建一个侦听 websocket 的 websocket。**sendMessage()** 函数将输入消息定向到 WebSocket URL。

此路由在应用程序代码中调用 **websocket_endpoint()** 函数。传入的连接请求被接受，传入的消息在客户端浏览器上回显。将以下代码添加到 main.py 中。

from fastapi import WebSocket
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
   await websocket.accept()
   while True:
      data = await websocket.receive_text()
      await websocket.send_text(f"Message text was: {data}")

保存 FastAPI 代码文件（main.py）、模板（socket.html）和 JavaScript 文件（ws.js）。运行 Uvicorn 服务器并访问 https://127.0.0.1:8000/ 以呈现如下所示的聊天窗口 -

键入某些文本并按下发送按钮。输入消息将通过 websocket 重定向到浏览器。

FastAPI - FastAPI 事件处理器

事件处理程序是在发生某些识别事件时要执行的函数。在 FastAPI 中，识别了两个这样的事件 **- 启动** 和 **关闭**。FastAPI 的应用程序对象具有 **on_event()** 装饰器，该装饰器使用这些事件之一作为参数。当发生相应的事件时，将触发使用此装饰器注册的函数。

启动事件发生在开发服务器启动之前，注册的函数通常用于执行某些初始化任务，例如与数据库建立连接等。关闭事件的事件处理程序在应用程序关闭之前立即调用。

示例

这是一个启动和关闭事件处理程序的简单示例。当应用程序启动时，启动时间将在控制台日志中回显。类似地，当通过按下 ctrl+c 停止服务器时，关闭时间也会显示。

main.py

from fastapi import FastAPI
import datetime
app = FastAPI()
@app.on_event("startup")
async def startup_event():
   print('Server started :', datetime.datetime.now())
@app.on_event("shutdown")
async def shutdown_event():
   print('server Shutdown :', datetime.datetime.now())

输出

它将产生以下输出 -

uvicorn main:app --reload
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [28720]
INFO: Started server process [28722]
INFO: Waiting for application startup.
Server started: 2021-11-23 23:51:45.907691
INFO: Application startup complete.
INFO: Shutting down
INFO: Waiting for application
server Shutdown: 2021-11-23 23:51:50.82955
INFO: Application shutdown com
INFO: Finished server process

FastAPI - 挂载子应用程序

如果您有两个独立的 FastAPI 应用程序，则其中一个可以安装在另一个应用程序之上。安装的应用程序称为子应用程序。**app.mount()** 方法在主应用程序的特定路径中添加另一个完全“独立”的应用程序。然后，它负责处理该路径下的所有内容，以及在该子应用程序中声明的路径操作。

让我们首先声明一个简单的 FastAPI 应用程序对象，用作顶级应用程序。

from fastapi import FastAPI
app = FastAPI()
@app.get("/app")
def mainindex():
   return {"message": "Hello World from Top level app"}

然后创建另一个应用程序对象 subapp 并添加它自己的路径操作。

subapp = FastAPI()
@subapp.get("/sub")
def subindex():
   return {"message": "Hello World from sub app"}

使用 mount() 方法将此 subapp 对象安装到主应用程序上。需要的两个参数是 URL 路由和子应用程序的名称。

app.mount("/subapp", subapp)

主应用程序和子应用程序都将拥有自己的文档，可以使用 Swagger UI 进行检查。

子应用程序的文档位于 https://127.0.0.1:8000/subapp/docs

FastAPI - 中间件

**中间件** 是一个函数，它与每个请求（在由任何特定路径操作处理之前）以及在返回每个响应之前处理。此函数接收发送到您的应用程序的每个请求。它可以通过运行其中定义的代码对请求执行某些处理，然后将请求传递给相应的操作函数进行处理。它还可以在返回之前处理操作函数生成的响应。

以下是 FastAPI 库中提供的一些中间件 -

• CORSMiddleware

• HTTPSRedirectMiddleware

• TrustedHostMiddleware

• GZipMiddleware

FastAPI 提供 **app.add_middleware()** 函数来处理服务器错误和自定义异常处理程序。除了上述集成的中间件之外，还可以定义自定义中间件。以下示例定义了 **addmiddleware()** 函数，并通过使用 **@app.middleware()** 装饰器将其装饰为中间件

该函数有两个参数，HTTP 请求对象和 **call_next()** 函数，该函数将向其相应的路径发送 API 请求并返回响应。

除了中间件函数之外，应用程序还有两个操作函数。

import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def addmiddleware(request: Request, call_next):
   print("Middleware works!")
   response = await call_next(request)
   return response
@app.get("/")
async def index():
   return {"message":"Hello World"}
@app.get("/{name}")
async def hello(name:str):
   return {"message":"Hello "+name}

当应用程序运行时，对于浏览器发出的每个请求，中间件输出（中间件有效！）将在响应输出之前出现在控制台日志中。

FastAPI - 挂载 Flask 应用程序

可以用 **WSGIMiddleware** 包装用 Flask 或 Django 框架编写的 WSGI 应用程序，并将其安装在 FastAPI 应用程序上以使其符合 ASGI。

首先在当前 FastAPI 环境中安装 Flask 包。

pip3 install flask

以下代码是最小的 Flask 应用程序 -

from flask import Flask
flask_app = Flask(__name__)
@flask_app.route("/")
def index_flask():
   return "Hello World from Flask!"

然后将 app 声明为 FastAPI 应用程序对象，并定义一个操作函数以呈现 Hello World 消息。

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

接下来，使用 mount() 方法将 flask 应用程序作为 FastAPI 主应用程序的子应用程序安装。

from fastapi.middleware.wsgi import WSGIMiddleware
app.mount("/flask", WSGIMiddleware(flask_app))

运行 Uvicorn 开发服务器。

uvicorn flaskapp:app –reload

主 FastAPI 应用程序位于 URL https://127.0.0.1:8000/ 路由。

{"message":"Hello World from FastAPI!"}

Flask 子应用程序安装在 URL https://127.0.0.1:8000/flask 上。

Hello World from Flask!

FastAPI - 部署

到目前为止，我们一直在使用本地开发服务器“Uvicorn”来运行我们的 FastAPI 应用程序。为了使应用程序公开可用，必须将其部署到具有静态 IP 地址的远程服务器上。可以使用免费计划或基于订阅的服务将其部署到不同的平台，例如 Heroku、Google Cloud、nginx 等。

在本章中，我们将使用 **Deta** 云平台。它的免费部署服务非常易于使用。

首先，要使用 Deta，我们需要在其网站上创建一个帐户，并使用您选择的合适的用户名和密码。

创建帐户后，在本地机器上安装 **Deta CLI**（命令行界面）。为您的应用程序创建一个文件夹（c:\fastapi_deta_app）如果您使用的是 Linux，请在终端中使用以下命令 -

iwr https://get.deta.dev/cli.ps1 -useb | iex

如果您使用的是 Windows，请从 Windows PowerShell 终端运行以下命令 -

PS C:\fastapi_deta_app> iwr https://get.deta.dev/cli.ps1 -useb | iex
Deta was installed successfully to
C:\Users\User\.deta\bin\deta.exe
Run 'deta --help' to get started

使用登录命令并验证您的用户名和密码。

PS C:\fastapi_deta_app> deta login
Please, log in from the web page. Waiting...
https://web.deta.sh/cli/60836
Logged in successfully.

在同一个应用程序文件夹中，在 **main.py** 文件中创建一个最小的 FastAPI 应用程序

# main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
   return {"Hello": "World"}
@app.get("/items/{item_id}")
def read_item(item_id: int):
   return {"item_id": item_id}

现在我们准备部署我们的应用程序。从 PowerShell 终端使用 **deta new** 命令。

PS C:\fastapi_deta_app> deta new
Successfully created a new micro
{
   "name": "fastapi_deta_app",
   "id": "2b236e8f-da6a-409b-8d51-7c3952157d3c",
   "project": "c03xflte",
   "runtime": "python3.9",
   "endpoint": "https://vfrjgd.deta.dev",
   "region": "ap-southeast-1",
   "visor": "enabled",
   "http_auth": "disabled"
}
Adding dependencies...
…..
Installing collected packages: typing-extensions, pydantic, idna, sniffio, anyio, starlette, fastapi
Successfully installed anyio-3.4.0 fastapi-0.70.0 idna-3.3 pydantic-1.8.2 sniffio-1.2.0 starlette-0.16.0 typingextensions-4.0.0

Deta 会将应用程序部署到指定的端点（对于每个应用程序，端点可能都是随机生成的）。它首先安装所需的依赖项，就像在本地机器上安装一样。部署成功后，打开浏览器并访问端点键前面显示的 URL。您也可以在 https://vfrigd.deta.dev/docs 找到 Swagger UI 文档。