FastAPI总结(一)
2025-03-19 19:40:19

FastAPI总结(一)

说明

在学习FastAPI的时候,我已经有了一些其他后端的使用经验,所以这篇笔记不是0基础开始的。

这部分总结分成一、二两部分:

  • 第一部分介绍FastAPI的路径、各种参数、依赖注入等关键技术细节。
  • 第二部分介绍安全性关系型数据库这两个重要的设计思想;还会介绍错误处理、中间件、后台任务、多文件项目、测试等辅助技术或设计

大纲

第一部分

  • 路径操作
  • 3种重要参数
  • 参数限定
  • Cookie, Header, Form, File等参数
  • 依赖注入

第二部分

  • 关系型数据库
  • 安全性
  • 多文件项目
  • 错误处理
  • 中间件
  • 后台任务
  • 其他杂项

简介

FastAPI 是一个用于构建 API 的轻量高性能web框架。

关键特性:

  • 快速:可与 NodeJSGo 并肩的极高性能(归功于 Starlette 和 Pydantic)。
  • 高效编码:提高功能开发速度约 200% 至 300%。
  • **并发:**Django和Flask默认的服务方式都是非并发,而FastAPI则良好支持并发。

Quick Start

安装

1
2
pip install fastapi //本体
pip install "uvicorn[standard]" // 服务器

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from typing import Union

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, q: Union[str, None] = None):
return {"item_id": item_id, "q": q}

运行

1
2
3
4
5
6
$ 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.

uvicorn main:app 命令含义如下:

  • mainmain.py 文件(一个 Python「模块」)。
  • app:在 main.py 文件中通过 app = FastAPI() 创建的对象。
  • --reload:让服务器在更新代码后重新启动。仅在开发时使用该选项。

路径操作

路径操作函数就是后端中常见的,以注解形式声明的,针对一个URL路径的处理操作。

路径装饰器

就是所谓的注解,例如@app.get("/")。显然,应该在双引号内填写这个路径操作函数针对的URL

1
2
3
4
5
6
from fastapi import FastAPI
app = FastAPI()

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

常见路径装饰器:

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

以及更少见的:

  • @app.options()
  • @app.head()
  • @app.patch()
  • @app.trace()

三种参数

在 FastAPI 中,参数的类型(是请求体参数还是查询参数)是由 参数的位置和声明方式 自动推断的。具体来说,FastAPI 会根据以下规则来判断参数的类型:

  1. 路径参数:写在URL中
1
2
3
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
  1. 查询参数:不在URL中,且非JSON格式
1
2
3
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
  1. 请求参数:不在URL中,且为JSON格式
1
2
3
4
5
6
7
8
9
from pydantic import BaseModel

class Item(BaseModel):
name: str
price: float

@app.post("/items/")
async def create_item(item: Item):
return item

后续,我们将详细讨论这3种参数的声明、限定等细节。

路径参数

即写在URL路径中的参数。

1
2
3
4
5
6
7
8
from fastapi import FastAPI

app = FastAPI()


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

顺序问题

如果有两个路径操作函数,他们有共同的前半段URL,但是在末端不一样:函数一的末端是一个单纯的路径;函数二的末端是一个路径参数。此时需要注意,将函数二声明在函数一之前,提高它的优先级。否则,当客户端访问函数一时,函数一的末端可能被视为一个路径参数。

1
2
3
4
5
6
7
8
9
10
11
12
13
from fastapi import FastAPI

app = 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}

文件路径作为路径参数

你需要这样做:

  1. 在路径参数变量名的后面加上:path
  2. 客户端访问URL时,在{file_path:path}的部分的前面多加一个斜杠,总共2个斜杠: /files//home/johndoe/myfile.txt
1
2
3
4
5
6
7
8
from fastapi import FastAPI

app = FastAPI()


@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
return {"file_path": file_path}

查询参数

路径操作函数可以接受客户端请求中位于请求体部分的数据,接收方法是将这些数据同名地声明为函数的参数,这样的参数就叫查询参数。

默认值

你还可以为这些查询参数指定默认值,当某个参数有默认值时,这个查询参数就具备可选性质。

1
2
3
4
5
6
7
8
9
10
from fastapi import FastAPI

app = 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]

多种类型可能的参数

你可以通过|或者Union将某个查询参数的类型声明为多种。

1
2
3
4
5
6
7
8
9
10
from fastapi import FastAPI

app = FastAPI()


@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}

请求参数

请求参数需要封装成类,因为他常常是多个键值对的JSON格式。这被称为请求参数模型

**注意:**这个封装类应当继承BaseModel

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from fastapi import FastAPI
from pydantic import BaseModel


class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None


app = FastAPI()


@app.post("/items/")
async def create_item(item: Item):
return item

查询参数模型

就像请求参数一样,查询参数也可以使用模型。然而,路径参数则几乎不使用参数模型。

注:在代码中,Field 是 Pydantic 提供的一个工具,用于为模型的字段(属性)定义额外的元数据或约束条件。它的作用是为字段提供更详细的配置,比如默认值、验证规则、描述信息等。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from typing import Annotated, Literal

from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI()


class FilterParams(BaseModel):
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []


@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
return filter_query

参数限定

也称“为参数提供额外的信息和校验”。

我们通过一些特殊的函数来实现,使得接收到的参数始终处于我们想要的样子。

Query:查询参数限定

这个函数用于限定请求参数

1
2
3
4
5
6
7
8
9
10
11
12
13
from typing import Union

from fastapi import FastAPI, Query

app = 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

可选限定

注意:以下限定描述可以用于任意一种参数

  • default:默认值,如果声明了默认值,则该参数为可选参数
  • max_length
  • min_length
  • pattern:正则表达式
  • gt:大于(greater than)
  • ge:大于等于(greater than or equal)
  • lt:小于(less than)
  • le:小于等于(less than or equal)

Annotated:依赖注入

这个关键字用于依赖注入,他的语法格式为:Annotated[<type>, <ProtoData>]

在这个小节,它可以用于向一个参数注入其类型限定

如果单纯从语法形式上学习他的用法,可以分为两种情况:

  1. 对参数使用依赖注入时,应当将这个结果看作是类型,也就是带有限定的类型
1
2
3
4
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,
):
  1. 否则,把他看作值,也就是带有限定的值,但是这个限定将永远存在(尽管具体的值可能会变)
1
Username = Annotated[str, Field(min_length=3, max_length=50)]

Path:路径参数限定

同理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from typing import Annotated

from fastapi import FastAPI, Path, Query

app = 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

请求参数限定

请求参数是自带限定的,他的Pydantic模型已经完成了此功能。

你还可以显式地使用Body来限定请求参数。

下面的例子中,对importance进行了显式声明。另外,这里的显式声明是必须的,否则路径操作函数将会把importance看作查询参数。

1
2
3
4
5
6
7
8
9
@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

对于上述路径操作函数,他期望这样一个请求:

1
2
3
4
5
6
7
8
9
10
11
12
13
{
"item": {
"name": "Foo",
"description": "The pretender",
"price": 42.0,
"tax": 3.2
},
"user": {
"username": "dave",
"full_name": "Dave Grohl"
},
"importance": 5
}

嵌入单个请求参数

如果只有一个请求参数,则不会有类似于上面”item”这样的键,如果我们需要一个这样的键,则要在Body中设置(embed=True)。例如:

路径操作函数:

1
2
3
4
@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

期望请求:

1
2
3
4
5
6
7
8
{
"item": {
"name": "Foo",
"description": "The pretender",
"price": 42.0,
"tax": 3.2
}
}

而不是:

1
2
3
4
5
6
{
"name": "Foo",
"description": "The pretender",
"price": 42.0,
"tax": 3.2
}

*号的使用

在 Python 中,函数参数的顺序是有规则的:没有默认值的参数(必需参数)必须放在 有默认值的参数(可选参数)之前。如果违反这个规则,Python 会报错。

而Path或者Query关键字会使得某个参数具有可选性质。当我们处理多个参数的时候,这一点让参数顺序的处理更麻烦。

FastAPI 提供了一种灵活的方式来解决这个问题:使用 * 作为函数的第一个参数。然后我们就可以无视参数顺序的规则。

参数的其他类型

以上,我们已经介绍了3种常见参数的用法,但目前这些介绍仍然局限于基本类型或者是由基本类型封装成的类模型。然而,以下这些类型也是常用的:

  • UUID:
    • 一种标准的 “通用唯一标识符” ,在许多数据库和系统中用作ID。
    • 在请求和响应中将以 str 表示。
  • datetime.datetime:
    • 一个 Python datetime.datetime.
    • 在请求和响应中将表示为 ISO 8601 格式的 str ,比如: 2008-09-15T15:53:00+05:00.
  • datetime.date:
    • Python datetime.date.
    • 在请求和响应中将表示为 ISO 8601 格式的 str ,比如: 2008-09-15.
  • datetime.time:
    • 一个 Python datetime.time.
    • 在请求和响应中将表示为 ISO 8601 格式的 str ,比如: 14:23:55.003.
  • datetime.timedelta:
    • 一个 Python datetime.timedelta.
    • 在请求和响应中将表示为 float 代表总秒数。
    • Pydantic 也允许将其表示为 “ISO 8601 时间差异编码”, 查看文档了解更多信息
  • frozenset:
    • 在请求和响应中,作为 set 对待:
      • 在请求中,列表将被读取,消除重复,并将其转换为一个 set
      • 在响应中 set 将被转换为 list
      • 产生的模式将指定那些 set 的值是唯一的 (使用 JSON 模式的 uniqueItems)。
  • bytes:
    • 标准的 Python bytes
    • 在请求和响应中被当作 str 处理。
    • 生成的模式将指定这个 strbinary “格式”。
  • Decimal:
    • 标准的 Python Decimal
    • 在请求和响应中被当做 float 一样处理。

Cookie参数和Header参数

现在我们继续介绍参数。

就跟前面同理,Cookie参数和Header参数都能够:被获取、被限定、被封装为模型、作为依赖注入。

Cookie声明:Cookie()

Header声明:Header()

1
2
3
4
5
6
7
8
# 使用Cookie
@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
return {"ads_id": ads_id}
# 使用 Header
@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):
return {"User-Agent": user_agent}

**注意自动转换:**Header大部分标准请求头用连字符分隔,即减号(-)。但是 user-agent 这样的变量在 Python 中是无效的。因此,默认情况下,Header 把参数名中的字符由下划线(_)改为连字符(-)来提取并存档请求头 。同时,HTTP 的请求头不区分大小写,可以使用 Python 标准样式(即 snake_case)进行声明。因此,可以像在 Python 代码中一样使用 user_agent ,无需把首字母大写为 User_Agent 等形式。如需禁用下划线自动转换为连字符,可以把 Header 的 convert_underscores 参数设置为 False。

响应模型

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

  • 进行响应模型声明后,如果实际返回响应模型与声明响应模型相比
    • 缺少可选字段:字段值会被设置为 None
    • 缺少必选字段:会抛出 ValidationError
    • 多余字段:会被自动过滤掉。
    • 类型不匹配:FastAPI 会尝试自动转换,失败时会抛出 ValidationError
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class UserIn(BaseModel):
username: str
password: str
email: EmailStr
full_name: str | None = None


class UserOut(BaseModel):
username: str
email: EmailStr
full_name: str | None = None


@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:
return user

以上的例子中,返回的userpassword将会被过滤掉。

表单参数

请求体既可以使用JSON作为数据格式,也可以使用JSON,在这一小节我们讨论前者。

Form参数数据格式示例:

1
2
3
4
POST /login/ HTTP/1.1
Content-Type: application/x-www-form-urlencoded

username=john_doe&password=secret

如何让 FastAPI 识别表单数据?

  • 使用 Form() 来声明表单字段。
  • 确保客户端发送的请求头中 Content-Type 是 application/x-www-form-urlencoded 或 multipart/form-data(包含文件)。
  • JSON的 Content-Type 是 application/json

FastAPI获取Form参数示例:

1
2
3
4
5
6
7
8
from fastapi import FastAPI, Form

app = FastAPI()


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

同理,你可以想象,Form参数跟其他的参数一样,可以:被获取、被限定、被封装为模型、作为依赖注入。

文件参数

你可以使用bytes或者UploadFile获取文件参数,使用前者会读取文件到内存,使用后者会保存文件到存储。

注意

  • 在请求中,文件参数的 Content-Type 是 multipart/form-data。
  • 你可以同时请求普通表单参数和文件参数,但是一旦使用了其中任意一种,就不能请求JSON格式的参数(即上面提到的请求参数)。
1
2
3
4
5
6
7
8
9
10
11
12
13
from fastapi import FastAPI, File, UploadFile

app = 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 = File()):
return {"filename": file.filename}

依赖注入

一个项的部分必须依靠另一个项得到实现,另一个项就成为依赖项,声明这种关系这一动作就是依赖注入。

更通俗地说:一个函数运行时,会自动调用另一个函数。这就是依赖。

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

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

函数依赖注入

**示例:**下面的例子展现了依赖注入在共享数据方面的作用。

read_items和read_users会调用common_parameters并获取相同结果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from typing import Union

from fastapi import Depends, FastAPI

app = 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

类依赖注入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from fastapi import Depends, FastAPI

app = 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 = q
self.skip = skip
self.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

你也可以这样写,这样的语法更简洁:

1
commons: CommonQueryParams = Depends()

多个依赖项

显而易见,一个依赖项也可以具有依赖项,且一个依赖项可以被使用多次。当存在一种类似于依赖路径的结构时,他们的调用顺序也是显而易见的,无需赘述。

没有返回值的依赖项

若没有返回值,你可以使用dependencies来进行依赖注入。

1
2
3
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
return [{"item": "Foo"}, {"item": "Bar"}]

全局依赖项

你可以通过给app注册依赖的形式,声明全局依赖项。这样一来,就可以为所有路径操作注入该依赖项。

1
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])

依赖注入配合yield

典型用法:

1
2
3
4
5
6
async def get_db():
db = DBSession()
try:
yield db
finally:
db.close()

此外,还可以配合HTTPException使用。

yield是什么?https://fastapi.tiangolo.com/zh/tutorial/dependencies/dependencies-with-yield/#_1