FastAPI 基础 - FastAPI essentails

  • 学习 py 后端,flask 太老,django 太重,就 FastAPI 了
  • Ref: Learn - FastAPI

Quick Setup

基于 uv

1
2
3
uv init awesome-project --bare
cd awesome-project
uv add "fastapi[standard]"

编辑 main.py

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

app = FastAPI()


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

运行项目,可以先配置 toml

pyproject.toml
1
2
[tool.fastapi]
entrypoint = "main:app"

然后

1
uv run fastapi dev

随后可以在 /docs 和 /redoc 查看自动生成的 API 文档。

Path Parameters

类似大多数路由框架,FastAPI 也支持路径参数。

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

还可以标注路径参数的类型,且当配置了类型时,框架会自动进行类型校验以及转换,例如从 str 到 int

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

路由顺序

例如你有两个路由:

1
2
3
4
5
6
7
8
@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/me 的路由在 /users/{user_id} 之前,否则 /users/me 会被匹配到 /users/{user_id},从而导致无法访问,同样的对于两个路径相同的路由,后一个会被前一个覆盖。

枚举类型

通过枚举类型可以限定路径参数的值,例如:

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

from fastapi import FastAPI


class 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"}

Path Parameters and Qeury Parameters at the same time

与 Query 类似

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

这里有一些细节,如果你不需要定义 query 的默认值,但需要给 path 添加默认值/验证规则,要么你使用 Annotated 以类型形式定义,要么你得把 path 之后的参数全都加上默认值,这里有一个 trick,可以用 * 放在签名的开始,让后续所有参数都转为 kwargs 于是就无需关注有无默认值的顺序。

Number validations

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

from fastapi import FastAPI, Path

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
item_id: Annotated[int, Path(title="The ID of the item to get", ge=1)], q: str
):
results = {"item_id": item_id}
if q:
results.update({"q": q})
return results

与 LaTeX 语法类似

  • ge - greater than or equal to
  • gt - greater than
  • le - less than or equal to
  • lt - less than

Query Parameters

任何非路径参数的参数都可以被认为是查询参数,例如:

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]

同样的,如果你在参数上标注了类型,那么框架也会自动进行类型校验和转换,包括 = 的默认值

Optional parameters

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}

通过 None 的默认值可以声明可选参数,反之若无默认值则为必选参数

Query Parameter Models

可以使用 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 Parameters and String Validations

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

from fastapi import FastAPI, Query

app = 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 来声明查询参数的复合类型(验证规则),例如上例中声明了 q 的最大长度为 50。

正则

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

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(
q: Annotated[
str | None, Query(min_length=3, max_length=50, pattern="^fixedquery$")
] = None,
):
results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
if q:
results.update({"q": q})
return results

默认值

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

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Annotated[str, Query(min_length=3)] = "fixedquery"):
results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
if q:
results.update({"q": q})
return results

列表查询参数

1
2
3
4
5
6
7
8
9
10
11
from typing import Annotated

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Annotated[list[str], Query()] = ["foo", "bar"]):
query_items = {"q": q}
return query_items

你必须用 Annotated 以及 Query 来显式声明列表类型的查询参数,否则会被解析为请求体。

查询参数别名

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

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Annotated[str | None, Query(alias="item-query")] = None):
results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
if q:
results.update({"q": q})
return results

对于类似 item-query 这样的查询参数,Python 中的变量名不能包含 -,所以需要通过 alias 来指定查询参数的别名。

这个 alias 不能理解为别名,因为这个情况下无法使用 q 传参

自定义校验

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import random
from typing import Annotated

from fastapi import FastAPI
from pydantic import AfterValidator

app = FastAPI()

data = {
"isbn-9781529046137": "The Hitchhiker's Guide to the Galaxy",
"imdb-tt0371724": "The Hitchhiker's Guide to the Galaxy",
"isbn-9781439512982": "Isaac Asimov: The Complete Stories, Vol. 2",
}


def check_valid_id(id: str):
if not id.startswith(("isbn-", "imdb-")):
raise ValueError('Invalid ID format, it must start with "isbn-" or "imdb-"')
return id


@app.get("/items/")
async def read_items(
id: Annotated[str | None, AfterValidator(check_valid_id)] = None,
):
if id:
item = data.get(id)
else:
id, item = random.choice(list(data.items()))
return {"id": id, "name": item}

这里 AfterValidator 可以在参数解析后(str 校验通过)进行自定义校验,若校验失败则会返回 422 错误。

如果需要在检查时使用外部组件,例如数据库,HTTP 请求,则需要使用 FastAPI 的依赖注入

Request Body

通过 Pydantic 来声明请求体

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

路由函数参数最终会按如下规则解析:

  • 如果一个参数它在路径中,那么它会被解析为路径参数
  • 如果一个参数是一个简单类型(str、int、float、bool),那么它会被解析为查询参数
  • 如果一个参数是一个 Pydantic 模型类的实例,那么它会被解析为请求体

单一参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from typing import Annotated

from fastapi import FastAPI, Path
from pydantic import BaseModel

app = FastAPI()


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


@app.put("/items/{item_id}")
async def update_item(
item_id: Annotated[int, Path(title="The ID of the item to get", ge=0, le=1000)],
q: str | None = None,
item: Item | None = None,
):
results = {"item_id": item_id}
if q:
results.update({"q": q})
if item:
results.update({"item": item})
return results

上面这个 example 期望一个如下形式的 JSON item

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

多个参数

当你有多个 Body Parameters 时

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

app = FastAPI()


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


class User(BaseModel):
username: str
full_name: str | None = None


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
results = {"item_id": item_id, "item": item, "user": user}
return results

这个例子期望如下形式的 JSON

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

通过 key 来区分不同的 Body Parameters,并根据变量名作为 key 从 JSON 中提取

基本类型的 Body

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


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


class User(BaseModel):
username: str
full_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

需要强制标识 Body(),与 Query()、Path()、Header()、Cookie() 一样,来告诉框架这是一个请求体参数,而不是查询参数或路径参数。

总结来说:

  • 参数名出现在路径中 → Path
  • 参数是简单类型(int、str、float 等) → Query
  • 参数是 Pydantic 模型 → Body

Fields

如果你需要参数校验,那么你需要用 Pydantic 模型进行而不是 FastAPI

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

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

app = FastAPI()


class Item(BaseModel):
name: str
description: str | None = Field(
default=None, title="The description of the item", max_length=300
)
price: float = Field(gt=0, description="The price must be greater than zero")
tax: 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

Nested Models

直接用 Pydantic 的嵌套模型即可

同样类似

1
2
3
4
5
6
7
8
9
10
from typing import Annotated

from fastapi import Cookie, FastAPI

app = FastAPI()


@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
return {"ads_id": ads_id}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from typing import Annotated

from fastapi import Cookie, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Cookies(BaseModel):
session_id: str
fatebook_tracker: str | None = None
googall_tracker: str | None = None


@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
return cookies

Header Parameters

类似处理

1
2
3
4
5
6
7
8
9
10
from typing import Annotated

from fastapi import FastAPI, Header

app = FastAPI()


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

Automatic conversion

即连字符的处理,例如 User-Agent 会被默认转换为 user_agent,也可以关闭

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

from fastapi import FastAPI, Header

app = FastAPI()


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

Header Parameter Models

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

from fastapi import FastAPI, Header
from pydantic import BaseModel

app = FastAPI()


class CommonHeaders(BaseModel):
host: str
save_data: bool
if_modified_since: str | None = None
traceparent: str | None = None
x_tag: list[str] = []


@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
return headers

Response Model - Return Type

通过用 Pydantic 模型标注返回类型,FastAPI 会自动将返回值转换为 JSON

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

app = FastAPI()


class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None
tags: list[str] = []


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


@app.get("/items/")
async def read_items() -> list[Item]:
return [
Item(name="Portal Gun", price=42.0),
Item(name="Plumbus", price=32.0),
]

response_model Parameter

response_model 是用来解决 return type 和 API 的 Output schema 不一致的问题,主要是转为一个更小属性类时做自动 filter 或者说转换用,比如下面这个例子,从 UserIn 转为 UserOut,从而隐藏了 password 字段。

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

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


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

不过自然的,也就可以用基类来实现

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

app = FastAPI()


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


class UserIn(BaseUser):
password: str


@app.post("/user/")
async def create_user(user: UserIn) -> BaseUser:
return user

Return a Response Directly

1
2
3
4
5
6
7
8
9
10
11
from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse, RedirectResponse

app = FastAPI()


@app.get("/portal")
async def get_portal(teleport: bool = False) -> Response:
if teleport:
return RedirectResponse(url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")
return JSONResponse(content={"message": "Here's your interdimensional portal."})

Response Status Code

这里是直接设置一个路由的默认返回状态码

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

app = FastAPI()


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

或者

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

app = FastAPI()


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

Form Data

To use forms, first install python-multipart. $ uv add python-multipart

1
2
3
4
5
6
7
8
9
10
from typing import Annotated

from fastapi import FastAPI, Form

app = FastAPI()


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

Form is a class that inherits directly from Body.

Form Models

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

from fastapi import FastAPI, Form
from pydantic import BaseModel

app = FastAPI()


class FormData(BaseModel):
username: str
password: str


@app.post("/login/")
async def login(data: Annotated[FormData, Form()]):
return data

Request Files

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

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/files/")
async def create_file(file: Annotated[bytes, File()]):
return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
return {"filename": file.filename}

通过 File() 来声明一个文件参数,会以 Form Data 的方式处理,这里声明了类型为 bytes 会以字节数组形式给出,这会使其存储与内存中,不适合打文件

而 UploadFile 是一个更高级的文件类型,支持异步读取、文件名、内容类型等,同时其采用了 spooling。

同样也可以做可选上传

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

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/files/")
async def create_file(file: Annotated[bytes | None, File()] = 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}

Multiple File Uploads

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from typing import Annotated

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse

app = FastAPI()


@app.post("/files/")
async def create_files(files: Annotated[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.get("/")
async def main():
content = """
<body>
<form action="/files/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
</body>
"""
return HTMLResponse(content=content)

这里的 List[bytes] 说明是多个文件,而 bytes 是单个文件,因为默认以 List 包装