FastAPI

Fast API 是一个快速、高效、简单、标准化的 Web 框架

基于 Python3.8+进行使用

Hello FastAPI

接下来新建一个项目带领进入 FastAPI 的世界

依赖包安装

  • pip install fastapi[all]
1
fastapi[all]

实现 api

创建一个hellofastapi.py文件

1
2
3
4
5
6
7
8
9
10
11
12
# -*- coding:utf-8 -*-
# Author: Zachary

from fastapi import FastAPI

app = FastAPI()


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

运行

需要使用 fastapi 提供的一个uvicornASGI 网关服务器来启动 api 服务

  • uvicorn hellofastapi:app --reload
1
2
3
4
5
6
7
(venv) ➜  fastApiProject uvicorn hellofastapi:app --reload
INFO:     Will watch for changes in these directories: ['/Users/zachary/Documents/PythonCode/fastApiProject']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [2378] using WatchFiles
INFO:     Started server process [2382]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

直接在浏览器输入下方的地址：http://127.0.0.1:8000

我们再额外加一个 api

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# -*- coding:utf-8 -*-
# Author: Zachary

from fastapi import FastAPI

app = FastAPI()


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


@app.get("/hello/{user}")
async def hello_user(user):
    return {"message": f"Hello {user}"}

文档

直接在浏览器键入：http://127.0.0.1:8000/docs 就有已经自动帮我们生成好的 swagger 文档

快速运行

当然也可以不使用命令行运行我们的 api 项目

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# -*- coding:utf-8 -*-
# Author: Zachary
import uvicorn
from fastapi import FastAPI

app = FastAPI()


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


@app.get("/hello/{user}")
async def hello_user(user):
    return {"message": f"Hello {user}"}


if __name__ == '__main__':
    uvicorn.run("hellofastapi:app", reload=True)

路径参数

路径参数在 URL 中

在 URL 中可以存在参数作为变量使用到程序中

重新建立一个目录/path_params在其中新建一个文件main.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# -*- coding:utf-8 -*-
# Author: Zachary
import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/user/{user_id}")
async def get_user(user_id):
    user = {
        "id": user_id,
        "name": "John Doe",
        "gender": "male"
    }
    return {"user": user}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

路径参数的校验

如果给路径参数指定类型，例如int，那么在进行 api 调用的时候 swagger 文档会自动帮我们进行一个校验

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# -*- coding:utf-8 -*-
# Author: Zachary
import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/user/{user_id}")
async def get_user(user_id: int):
    user = {
        "id": user_id,
        "name": "John Doe",
        "gender": "male"
    }
    return {"user": user}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

api 定义的顺序

假定我现在需要访问这么两个 api：/users/current和/users/1

分别用于获取当前用户和 user_id 为 1 的用户

那么定义的时候顺序是很重要的

  • 第一种顺序
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
# -*- coding:utf-8 -*-
# Author: Zachary
import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = {
        "id": user_id,
        "name": "John Doe",
        "gender": "male"
    }
    return {"user": user}


@app.get("/users/current")
async def get_current_user():
    user = {
        "id": 2,
        "name": "Jane Guilherme",
        "gender": "female"
    }
    return {"user": user}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

访问/users/current会出问题

  • 第二种顺序

需要调换一下 current 和 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
24
25
26
27
28
29
30
# -*- coding:utf-8 -*-
# Author: Zachary
import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/users/current")
async def get_current_user():
    user = {
        "id": 2,
        "name": "Jane Guilherme",
        "gender": "female"
    }
    return {"user": user}


@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = {
        "id": user_id,
        "name": "John Doe",
        "gender": "male"
    }
    return {"user": user}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

这样都可以成功访问到

枚举定义选择项

目前在 swagger-UI 中的一些参数都是需要手动键入的

但有时候一些参数可能是一些固定选项，例如性别（男、女）这时候如果能实现选择项输入就更好了

使用枚举类实现选择项的定义

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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum

import uvicorn
from fastapi import FastAPI

app = FastAPI()


class Gender(str, Enum):
    male = "男性"
    female = "女性"


@app.get("/students/{gender}")
async def get_student(gender: Gender):
    student = {
        "id": 1,
        "name": "Tony Stark",
        "gender": gender
    }
    return {"student": student}


@app.get("/users/current")
async def get_current_user():
    user = {
        "id": 2,
        "name": "Jane Guilherme",
        "gender": "female"
    }
    return {"user": user}


@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = {
        "id": user_id,
        "name": "John Doe",
        "gender": "male"
    }
    return {"user": user}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

查询参数

什么是查询参数，类似于http://127.0.0.1:8000/users?page_index=1&page_size=10

其中的 page_index 和 page_size 都是查询参数

来个实例看看怎么做，先创建一个/query_params目录，然后目录下面创建main.py

查询参数在方法参数中

  • 在注解处 就可以不用声明这两个查询参数了
  • 直接在方法的参数中声明
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
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum

import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/users")
async def get_current_user(page_index: int, page_size: int):
    user = {
        "id": 2,
        "name": "Jane Guilherme",
        "gender": "female"
    }
    page_info = {
        "page_index": page_index,
        "page_size": page_size
    }
    return {"user": user, "page_info": page_info}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

运行代码后直接在浏览器键入：http://127.0.0.1:8000/users?page_index=2&page_size=10

可以查看到结果

查询参数的校验

  • 同样，在方法中的类型声明，会自动帮我们做类型验证，如果输入的类型与声明类型不符，会报错

如果键入：http://127.0.0.1:8000/users?page_index=2&page_size=a

使用 swagger-UI 也会帮我们做参数类型校验

配置可选查询参数

  • 使用Optional声明可选查询参数，若查询参数不传值，希望有一个默认值
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
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/users")
async def get_current_user(page_index: int, page_size: Optional[int] = 10):
    user = {
        "id": 2,
        "name": "Jane Guilherme",
        "gender": "female"
    }
    page_info = {
        "page_index": page_index,
        "page_size": page_size
    }
    return {"user": user, "page_info": page_info}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

路径参数与查询参数混用

结合上一节的路径参数，可以与查询参数混合使用，并且在方法参数中的顺序是没有要求的

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
34
35
36
37
38
39
40
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/users")
async def get_current_user(page_index: int, page_size: Optional[int] = 10):
    user = {
        "id": 2,
        "name": "Jane Guilherme",
        "gender": "female"
    }
    page_info = {
        "page_index": page_index,
        "page_size": page_size
    }
    return {"user": user, "page_info": page_info}


@app.get("/users/{user_id}/friends")
async def get_user_friends(user_id: int, page_index: int, page_size: Optional[int] = 10):
    user_friends = {
        "id": user_id + 1,
        "name": "John Doe",
        "gender": "male"
    }
    page_info = {
        "page_index": page_index,
        "page_size": page_size
    }
    return {"user_friends": user_friends, "page_info": page_info}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

请求体

什么是请求体

请求体是用于发送数据给 API

  • 不能使用GET请求发送请求体
  • 发送请求体的类型应该是POST、PUT、DELETE或者PATCH

定义请求体中数据的模型

  • 使用pydantic模块下的BaseModel

在项目目录下创建一个/request_body目录，然后创建一个main.py

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
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class UserModel(BaseModel):
    id: Optional[int] = 1
    name: str
    description: Optional[str] = "No description"


@app.post("/users")
async def create_user(user: UserModel):
    # 做一些创建用户的数据库操作
    user = user.model_dump()
    return user


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

只传入非 Optional 的参数：

给可选参数传入值

不给非可选参数传值，会报错，指出缺少的参数信息

请求体中定义枚举选择项

在前面的路径参数中，对一些参数可以使用选择项进行输入，那么同样在请求体中的某些参数也可以配置为枚举类进行选择项输入

  • 枚举类定义选择项
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
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum
from typing import Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Gender(str, Enum):
    male = "男性"
    female = "女性"


class UserModel(BaseModel):
    id: Optional[int] = 1
    name: str
    gender: Gender
    description: Optional[str] = "No description"


@app.post("/users")
async def create_user(user: UserModel):
    # 做一些创建用户的数据库操作
    user = user.model_dump()
    return user


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

  • 会对枚举选择项，自动做内容校验

输入错误的枚举类型，结果会报错，具体错误信息

请求体与路径参数混用

这块逻辑需要先改一下

因为对于更新数据而言，我们可能请求体发送的时候只是发送要更新的数据，我们一般不会知道 id 是多少，所以 UserModel 先去掉 id，然后在修改完之后给他从路径参数中把 id 加过来

实际上在开发的时候这些 User 类就要做一下区分：如 UserVO，User，UserDTO 等等

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
34
35
36
37
38
39
40
41
42
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum
from typing import Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Gender(str, Enum):
    male = "男性"
    female = "女性"


class UserModel(BaseModel):
    name: str
    gender: Gender
    description: Optional[str] = "No description"


@app.post("/users")
async def create_user(user: UserModel):
    # 做一些创建用户的数据库操作
    user = user.model_dump()
    user.update({"id": 1})

    return user


@app.put("/users/{user_id}")
async def update_user(user_id: int, user: UserModel):
    # 做一些更新用户的数据库操作
    user = user.model_dump()
    user.update({"id": user_id})
    return user


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

函数的参数识别规则

  • FastAPI 如何区分方法中的路径参数、查询参数、请求体对象
    • 如果在路径参数中定义了，函数的参数会匹配为路径参数
    • 如果路径参数中没有定义，然后函数的参数定义为 int、str 等基础类型，会匹配为查询参数
    • 如果是 pydantic 模型类（继承 BaseModel），则匹配为请求体

参数验证

目录路径下创建一个/params_validation目录，然后创建一个main.py

路径参数的验证

  • 使用fastapi.Path类实现
  • 路径参数都是必须项
  • 格式：
    • ...表示该路径参数必填
    • title表示对该路径参数的名称
    • ge值大于等于，gt大于
    • le值小于等于，lt小于
    • min_length 字符串最小长度
    • max_length 字符串最大长度
    • regex正则表达式验证 常用格式r"^[正则内容]$"
1
2
3
4
5
@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(...,
                                       title="The ID of the user to get",
                                       ge=1,
                                       le=200)):

下面看一下具体实例

整数类型验证

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# -*- coding:utf-8 -*-
# Author: Zachary

import uvicorn
from fastapi import FastAPI, Path

app = FastAPI()


@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(..., title="The ID of the user to get", ge=1, le=200)):
    user = {
        "id": user_id,
        "name": "John Doe",
        "gender": "male"
    }
    return {"user": user}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

路径参数会显示最大值最小值范围，若输入不在范围，则会报错

上面这个例子是一个整数的类型，下面来一个字符串类型的验证

字符串类型验证

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
# -*- coding:utf-8 -*-
# Author: Zachary

import uvicorn
from fastapi import FastAPI, Path

app = FastAPI()


@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(..., title="The ID of the user to get", ge=1, le=200)):
    user = {
        "id": user_id,
        "name": "John Doe",
        "gender": "male"
    }
    return {"user": user}


@app.get("/books/{book_name}")
async def get_book(book_name: str = Path(..., title="The name of the book to get", min_length=3, max_length=10)):
    book = {
        "id": 1,
        "name": book_name,
        "description": "The description of the book..."
    }
    return {"book": book}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

正则表达式验证

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
34
35
36
37
38
39
40
41
42
# -*- coding:utf-8 -*-
# Author: Zachary

import uvicorn
from fastapi import FastAPI, Path

app = FastAPI()


@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(..., title="The ID of the user to get", ge=1, le=200)):
    user = {
        "id": user_id,
        "name": "John Doe",
        "gender": "male"
    }
    return {"user": user}


@app.get("/books/{book_name}")
async def get_book(book_name: str = Path(..., title="The name of the book to get", min_length=3, max_length=10)):
    book = {
        "id": 1,
        "name": book_name,
        "description": "The description of the book..."
    }
    return {"book": book}


@app.get("/milkteas/{milktea_name}")
async def get_milktea(
        milktea_name: str = Path(..., title="The name of the milktea to get", regex=r"^.*茶$", min_length=2)):
    milktea = {
        "id": 1,
        "name": milktea_name,
        "description": "The description of the milktea..."
    }
    return {"milktea": milktea}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

实现了一个查询奶茶的 api，其中要求奶茶名必须是茶结尾，并且字符串长度大于等于 2

查询参数的验证

  • 使用fastapi.Query类实现
  • 使用方法与Path大体一致
  • 区别在于：
    • Query(..., )表示该查询参数为必选项
    • Query(None, )表示该查询参数为 可选项，其中None可以是任何内容表示默认值
    • Query(alias="[别名]")设置查询参数别名，起了别名之后原先参数名失效

整数类型验证

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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
# -*- coding:utf-8 -*-
# Author: Zachary

import uvicorn
from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/users/current")
async def get_current_user(page_index: int = Query(1, ge=1, le=100),
                           page_size: int = Query(10, ge=1, le=100)):
    user = {
        "id": 2,
        "name": "Jane Guilherme",
        "gender": "female"
    }
    page_info = {
        "page_index": page_index,
        "page_size": page_size
    }
    return {"user": user, "page_info": page_info}


@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(..., title="The ID of the user to get", ge=1, le=200)):
    user = {
        "id": user_id,
        "name": "John Doe",
        "gender": "male"
    }
    return {"user": user}


@app.get("/books/{book_name}")
async def get_book(book_name: str = Path(..., title="The name of the book to get", min_length=3, max_length=10)):
    book = {
        "id": 1,
        "name": book_name,
        "description": "The description of the book..."
    }
    return {"book": book}


@app.get("/milkteas/{milktea_name}")
async def get_milktea(
        milktea_name: str = Path(..., title="The name of the milktea to get", regex=r"^.*茶$", min_length=2)):
    milktea = {
        "id": 1,
        "name": milktea_name,
        "description": "The description of the milktea..."
    }
    return {"milktea": milktea}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

查询参数别名

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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
# -*- coding:utf-8 -*-
# Author: Zachary

import uvicorn
from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/users/current")
async def get_current_user(page_index: int = Query(1, alias="page-index", ge=1, le=100),
                           page_size: int = Query(10, alias="page-size", ge=1, le=100)):
    user = {
        "id": 2,
        "name": "Jane Guilherme",
        "gender": "female"
    }
    page_info = {
        "page_index": page_index,
        "page_size": page_size
    }
    return {"user": user, "page_info": page_info}


@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(..., title="The ID of the user to get", ge=1, le=200)):
    user = {
        "id": user_id,
        "name": "John Doe",
        "gender": "male"
    }
    return {"user": user}


@app.get("/books/{book_name}")
async def get_book(book_name: str = Path(..., title="The name of the book to get", min_length=3, max_length=10)):
    book = {
        "id": 1,
        "name": book_name,
        "description": "The description of the book..."
    }
    return {"book": book}


@app.get("/milkteas/{milktea_name}")
async def get_milktea(
        milktea_name: str = Path(..., title="The name of the milktea to get", regex=r"^.*茶$", min_length=2)):
    milktea = {
        "id": 1,
        "name": milktea_name,
        "description": "The description of the milktea..."
    }
    return {"milktea": milktea}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

起了查询参数别名之后，可以键入：http://127.0.0.1:8000/users/current?page-index=2&page-size=10

这时如果还使用原先的查询参数就会失效啦，会使用默认值而不是传入的值

请求体-进阶

创建一个/request_body_advance目录，然后创建一个main.py

多个请求体参数

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
34
35
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum
from typing import Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class User(BaseModel):
    name: str
    description: Optional[str] = "No description"


class Order(BaseModel):
    number: int
    goods: str


@app.put("/carts/{cart_id}")
async def update_cart(cart_id: int, user: User, order: Order):
    # 做一些更新购物车的数据库操作
    result = {
        "cart_id": cart_id,
        "user_name": user.name,
        "order_good": order.goods
    }
    return result


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

单一数据类型作为请求体参数

当函数参数列表中有一个 int 或者 str 类型的单一数据类型时，API 会默认将它作为一个查询参数，那么我们应该怎么做让其声明为一个请求体参数呢

  • 使用fastapi.Body类声明，使用方法与上面的 Path、Query 大体相同
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
34
35
36
37
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum
from typing import Optional

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

app = FastAPI()


class User(BaseModel):
    name: str
    description: Optional[str] = "No description"


class Order(BaseModel):
    number: int
    goods: str


@app.put("/carts/{cart_id}")
async def update_cart(cart_id: int, user: User, order: Order,
                      total_price: float = Body(...)):
    # 做一些更新购物车的数据库操作
    result = {
        "cart_id": cart_id,
        "user_name": user.name,
        "order_good": order.goods,
        "total_price": total_price
    }
    return result


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

这时 total_price 可以作为请求体参数发给 API

请求体模型中的属性验证

  • 使用pydantic.Field进行验证
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
34
35
36
37
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum
from typing import Optional

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

app = FastAPI()


class User(BaseModel):
    name: str = Field(..., min_length=4)
    description: Optional[str] = Field("No description", min_length=10)


class Order(BaseModel):
    number: int = Field(..., ge=1)
    goods: str


@app.put("/carts/{cart_id}")
async def update_cart(cart_id: int, user: User, order: Order,
                      total_price: float = Body(...)):
    # 做一些更新购物车的数据库操作
    result = {
        "cart_id": cart_id,
        "user_name": user.name,
        "order_good": order.goods,
        "total_price": total_price
    }
    return result


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

错误的参数：

正确的参数：

请求体模型嵌套

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
34
35
36
37
38
39
40
41
42
43
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum
from typing import Optional

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

app = FastAPI()


class Address(BaseModel):
    province: str
    city: str


class User(BaseModel):
    name: str = Field(..., min_length=4)
    description: Optional[str] = Field("No description", min_length=10)


class Order(BaseModel):
    number: int = Field(..., ge=1)
    goods: str
    address: Address


@app.put("/carts/{cart_id}")
async def update_cart(cart_id: int, user: User, order: Order,
                      total_price: float = Body(...)):
    # 做一些更新购物车的数据库操作
    result = {
        "cart_id": cart_id,
        "user_name": user.name,
        "order_good": order.goods,
        "total_price": total_price
    }
    return result


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

模型内使用 list、set 等数据结构

  • goods: List[str]
  • goods: list
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
34
35
36
37
38
39
40
41
42
43
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum
from typing import Optional, List

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

app = FastAPI()


class Address(BaseModel):
    province: str
    city: str


class User(BaseModel):
    name: str = Field(..., min_length=4)
    description: Optional[str] = Field("No description", min_length=10)


class Order(BaseModel):
    number: int = Field(..., ge=1)
    goods: List[str] = Field(..., min_items=1)
    address: Address


@app.put("/carts/{cart_id}")
async def update_cart(cart_id: int, user: User, order: Order,
                      total_price: float = Body(...)):
    # 做一些更新购物车的数据库操作
    result = {
        "cart_id": cart_id,
        "user_name": user.name,
        "order_good": order.goods,
        "total_price": total_price
    }
    return result


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

示例数据

在 swagger-UI 中每一个 API 会有一个默认的示例数据

示例数据是显示在文档中的例子，可以给使用者提供示例数据格式，可视化更直接，更便于使用者理解 API 如何使用

Body > model_config > Field

但是更推荐使用Field，因为如果请求体数据模型有修改，会更便利

通过 Field 定义示例数据

  • Field(..., examples=["具体数据"])
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
34
35
36
37
38
39
40
41
42
43
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum
from typing import Optional, List

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

app = FastAPI()


class Address(BaseModel):
    province: str = Field(..., examples=["广东省"])
    city: str = Field(..., examples=["深圳市"])


class User(BaseModel):
    name: str = Field(..., min_length=4, examples=["Zachary"])
    description: Optional[str] = Field("No description", min_length=10, examples=["该用户是新人，拥有新人优惠券"])


class Order(BaseModel):
    number: int = Field(..., ge=1, examples=[1])
    goods: List[str] = Field(..., min_items=1, examples=[["苹果", "香蕉"]])
    address: Address


@app.put("/carts/{cart_id}")
async def update_cart(cart_id: int, user: User, order: Order,
                      total_price: float = Body(...)):
    # 做一些更新购物车的数据库操作
    result = {
        "cart_id": cart_id,
        "user_name": user.name,
        "order_good": order.goods,
        "total_price": total_price
    }
    return result


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

通过定义模型属性 model_config

  • model_config优先级更高
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum
from typing import Optional, List

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

app = FastAPI()


class Address(BaseModel):
    province: str = Field(...)
    city: str = Field(...)
    model_config = {
        "json_schema_extra": {
            "examples": [{
                "province": "江苏省",
                "city": "南京市"
            }]
        }
    }


class User(BaseModel):
    name: str = Field(..., min_length=4)
    description: Optional[str] = Field("No description", min_length=10)
    model_config = {
        "json_schema_extra": {
            "examples": [{
                "name": "Zachary",
                "description": "该用户是新人，拥有新人优惠券"
            }]
        }
    }


class Order(BaseModel):
    number: int = Field(..., ge=1, examples=[1])
    goods: List[str] = Field(..., min_items=1, examples=[["苹果", "香蕉"]])
    address: Address


@app.put("/carts/{cart_id}")
async def update_cart(cart_id: int, user: User, order: Order,
                      total_price: float = Body(...)):
    # 做一些更新购物车的数据库操作
    result = {
        "cart_id": cart_id,
        "user_name": user.name,
        "order_good": order.goods,
        "total_price": total_price
    }
    return result


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

通过 Body 定义示例数据

  • Body主要是定义单一数据类型作为请求体参数时用的，但同时请求体数据模型也适用Body进行配置
  • 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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
# -*- coding:utf-8 -*-
# Author: Zachary
from enum import Enum
from typing import Optional, List

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

app = FastAPI()


class Address(BaseModel):
    province: str = Field(...)
    city: str = Field(...)
    model_config = {
        "json_schema_extra": {
            "examples": [{
                "province": "江苏省",
                "city": "南京市"
            }]
        }
    }


class User(BaseModel):
    name: str = Field(..., min_length=4)
    description: Optional[str] = Field("No description", min_length=10)
    model_config = {
        "json_schema_extra": {
            "examples": [{
                "name": "Zachary",
                "description": "该用户是新人，拥有新人优惠券"
            }]
        }
    }


class Order(BaseModel):
    number: int = Field(..., ge=1, examples=[1])
    goods: List[str] = Field(..., min_items=1, examples=[["苹果", "香蕉"]])
    address: Address


@app.put("/carts/{cart_id}")
async def update_cart(cart_id: int, user: User, order: Order,
                      total_price: float = Body(..., examples=[188888.88])):
    # 做一些更新购物车的数据库操作
    result = {
        "cart_id": cart_id,
        "user_name": user.name,
        "order_good": order.goods,
        "total_price": total_price
    }
    return result


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

这里提及一个小技巧

  • 对方法来说，如果def func(*, num1, num2)参数列表的第一个*把所有位置参数占了，后面的所有参数都是关键字参数，并且没有顺序

Cookie 与 Header 参数

创建一个目录/cookie_header，然后在这其中创建一个main.py

  • 放在 cookie 中的键值对数据，常用的键不使用_连接，常用-，但是用-不符合 Python 编程规范，因此可以设置别名
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional, Union

import uvicorn
from fastapi import FastAPI, Cookie, Header

app = FastAPI()


@app.put("/cookieAndHeader")
async def cookie_and_header(*,
                            favorite_schema: Optional[str] = Cookie(default=None, alias="favorite-schema"),
                            api_token: Union[str, None] = Header(default=None, alias="api-token")):
    result_dict = {
        "favorite_schema": favorite_schema,
        "api_token": api_token
    }
    return result_dict


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

但是尝试测试可以发现，在 swagger-UI 里面无法显示 cookie 的内容

这是由于浏览器的原因，需要使用 postman 来测试，设置好 header 和 cookie

这时，可以获得完整的 cookie

如果一定想要通过 swagger-UI 来实现 cookie 的查看，需要设置一下 cookie

  • 通过Response来设置 cookie 的内容，实际上这个 Response 就是发送给客户端结果的
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
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional, Union

import uvicorn
from fastapi import FastAPI, Cookie, Header, Response

app = FastAPI()


@app.put("/cookieAndHeader")
async def cookie_and_header(*,
                            response: Response,
                            favorite_schema: Optional[str] = Cookie(default=None, alias="favorite-schema"),
                            api_token: Union[str, None] = Header(default=None, alias="api-token")):
    result_dict = {
        "favorite_schema": favorite_schema,
        "api_token": api_token
    }

    response.set_cookie(key="favorite-schema", value="black")
    return result_dict


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

之后再去 swagger-UI 查看的时候，第一次调用仍旧没有 cookie，但是服务端进行了设置，第二次调用的时候，就能看到服务端所设置的 cookie 了

响应模型

在 swagger-UI 中，我们只能得到一个响应模型的精简例子，只能知道返回的内容是什么类型，不能知道一个具体格式

在项目路径下创建一个/response_model，然后创建一个main.py

定义响应模型

  • response_model指定响应模型
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
34
35
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

users = {
    "s": {"id": 0},
    "a": {"id": 1, "username": "a"},
    "b": {"id": 2, "username": "b", "password": "bbb"},
    "c": {"id": 3, "username": "c", "password": "ccc", "description": "default"},
    "d": {"id": 4, "username": "d", "password": "ddd", "description": "it's user d"},
    "e": {"id": 5, "username": "e", "password": "eee", "description": "it's user e", "fullname": "Tony Stark"}
}

app = FastAPI()


class UserOut(BaseModel):
    id: int
    username: str
    password: Optional[str] = "default"
    description: Optional[str] = "default"
    fullname: Optional[str] = "default"


@app.get("/users/{username}", response_model=UserOut)
async def get_user(username: str):
    return users.get(username, {})


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

这时查看一下 responses，就能看出来返回结果具体包括的内容

  • resoinse_model_include可以指定响应模型只输出的内容
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
34
35
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

users = {
    "s": {"id": 0},
    "a": {"id": 1, "username": "a"},
    "b": {"id": 2, "username": "b", "password": "bbb"},
    "c": {"id": 3, "username": "c", "password": "ccc", "description": "default"},
    "d": {"id": 4, "username": "d", "password": "ddd", "description": "it's user d"},
    "e": {"id": 5, "username": "e", "password": "eee", "description": "it's user e", "fullname": "Tony Stark"}
}

app = FastAPI()


class UserOut(BaseModel):
    id: int
    username: str
    password: Optional[str] = "default"
    description: Optional[str] = "default"
    fullname: Optional[str] = "default"


@app.get("/users/{username}", response_model=UserOut, response_model_include={"id", "username", "description"})
async def get_user(username: str):
    return users.get(username, {})


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

这样的话 response 只有指定的

  • response_model_exclude也是同样的用法
  • response_model_exclude_unset如果数据包含这个值，就显示出来
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
34
35
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

users = {
    "s": {"id": 0},
    "a": {"id": 1, "username": "a"},
    "b": {"id": 2, "username": "b", "password": "bbb"},
    "c": {"id": 3, "username": "c", "password": "ccc", "description": "default"},
    "d": {"id": 4, "username": "d", "password": "ddd", "description": "it's user d"},
    "e": {"id": 5, "username": "e", "password": "eee", "description": "it's user e", "fullname": "Tony Stark"}
}

app = FastAPI()


class UserOut(BaseModel):
    id: int
    username: str
    password: Optional[str] = "default"
    description: Optional[str] = "default"
    fullname: Optional[str] = "default"


@app.get("/users/{username}", response_model=UserOut, response_model_exclude_unset=True)
async def get_user(username: str):
    return users.get(username, {})


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

响应模型列表

  • response_model=List[User]
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
34
35
36
37
38
39
40
41
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional, List

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

users = {
    # "s": {"id": 0},
    "a": {"id": 1, "username": "a"},
    "b": {"id": 2, "username": "b", "password": "bbb"},
    "c": {"id": 3, "username": "c", "password": "ccc", "description": "default"},
    "d": {"id": 4, "username": "d", "password": "ddd", "description": "it's user d"},
    "e": {"id": 5, "username": "e", "password": "eee", "description": "it's user e", "fullname": "Tony Stark"}
}

app = FastAPI()


class UserOut(BaseModel):
    id: int
    username: str
    password: Optional[str] = "default"
    description: Optional[str] = "default"
    fullname: Optional[str] = "default"


@app.get("/users/{username}", response_model=UserOut, response_model_exclude_unset=True)
async def get_user(username: str):
    return users.get(username, {})


@app.get("/users", response_model=List[UserOut])
async def get_users():
    return users.values()


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

状态码与异常处理

创建一个/exception_handler目录，然后创建main.py

状态码

返回指定状态码

  • status_code=
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
34
35
36
37
38
39
40
41
42
43
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

users = {
    "a": {"id": 1, "username": "a"},
    "b": {"id": 2, "username": "b", "password": "bbb"},
    "c": {"id": 3, "username": "c", "password": "ccc", "description": "default"},
    "d": {"id": 4, "username": "d", "password": "ddd", "description": "it's user d"},
    "e": {"id": 5, "username": "e", "password": "eee", "description": "it's user e", "fullname": "Tony Stark"}
}


class UserBase(BaseModel):
    id: Optional[int] = None
    username: str
    description: Optional[str] = "default"
    fullname: Optional[str] = None


class UserIn(UserBase):
    password: str


class UserOut(UserBase):
    pass


@app.post("/users", status_code=201, response_model=UserOut)
async def create_user(user: UserIn):
    user_dict = user.model_dump()
    user_dict.update({"id": 6})
    return user_dict


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

异常处理

通过异常返回错误

  • 通过抛出HTTPException来获得错误返回
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI, Path, HTTPException, status
from pydantic import BaseModel

app = FastAPI()

users = {
    "a": {"id": 1, "username": "a"},
    "b": {"id": 2, "username": "b", "password": "bbb"},
    "c": {"id": 3, "username": "c", "password": "ccc", "description": "default"},
    "d": {"id": 4, "username": "d", "password": "ddd", "description": "it's user d"},
    "e": {"id": 5, "username": "e", "password": "eee", "description": "it's user e", "fullname": "Tony Stark"}
}


class UserBase(BaseModel):
    id: Optional[int] = None
    username: str
    description: Optional[str] = "default"
    fullname: Optional[str] = None


class UserIn(UserBase):
    password: str


class UserOut(UserBase):
    pass


@app.post("/users", status_code=201, response_model=UserOut)
async def create_user(user: UserIn):
    user_dict = user.model_dump()
    user_dict.update({"id": 6})
    return user_dict


@app.get("/users/{username}", status_code=200, response_model=UserOut)
async def get_user(username: str = Path(..., min_length=1)):
    user = users.get(username, None)
    if user:
        return user
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"{username} not found")


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

自定义异常

  • 通过使用@app.exception_handler()进行异常处理
  • 通过JSONResponse返回异常结果
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI, Path, HTTPException, status, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

users = {
    "a": {"id": 1, "username": "a"},
    "b": {"id": 2, "username": "b", "password": "bbb"},
    "c": {"id": 3, "username": "c", "password": "ccc", "description": "default"},
    "d": {"id": 4, "username": "d", "password": "ddd", "description": "it's user d"},
    "e": {"id": 5, "username": "e", "password": "eee", "description": "it's user e", "fullname": "Tony Stark"}
}


class UserBase(BaseModel):
    id: Optional[int] = None
    username: str
    description: Optional[str] = "default"
    fullname: Optional[str] = None


class UserIn(UserBase):
    password: str


class UserOut(UserBase):
    pass


class UserNotFoundException(Exception):
    def __init__(self, username: str):
        self.username = username


@app.post("/users", status_code=201, response_model=UserOut)
async def create_user(user: UserIn):
    user_dict = user.model_dump()
    user_dict.update({"id": 6})
    return user_dict


@app.get("/users/{username}", status_code=200, response_model=UserOut)
async def get_user(username: str = Path(..., min_length=1)):
    user = users.get(username, None)
    if user:
        return user
    raise UserNotFoundException(username)


@app.exception_handler(UserNotFoundException)
async def user_not_found_exception_handler(request: Request, exc: UserNotFoundException):
    return JSONResponse(status_code=404,
                        content={
                            "error_code": 404,
                            "message": f"user: {exc.username} not found"})


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

处理多种返回

  • 通过responses=参数指定会包含的多种异常返回响应模型
    • 但是这块需要注意一点：responses 中的异常响应模型，并不会对 return 的内容做格式化，所以需要自己处理一下返回的内容
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI, Path, HTTPException, status, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

users = {
    "a": {"id": 1, "username": "a"},
    "b": {"id": 2, "username": "b", "password": "bbb"},
    "c": {"id": 3, "username": "c", "password": "ccc", "description": "default"},
    "d": {"id": 4, "username": "d", "password": "ddd", "description": "it's user d"},
    "e": {"id": 5, "username": "e", "password": "eee", "description": "it's user e", "fullname": "Tony Stark"}
}


class UserBase(BaseModel):
    id: Optional[int] = None
    username: str
    description: Optional[str] = "default"
    fullname: Optional[str] = None


class UserIn(UserBase):
    password: str


class UserOut(UserBase):
    pass


class ErrorMessage(BaseModel):
    error_code: int
    message: str


class UserNotFoundException(Exception):
    def __init__(self, username: str):
        self.username = username


@app.post("/users", status_code=201, response_model=UserOut, responses={
    400: {
        "model": ErrorMessage,
    },
    401: {
        "model": ErrorMessage,
    }
})
async def create_user(user: UserIn):
    if users.get(user.username, None):
        error_message = ErrorMessage(error_code=400, message=f"user: {user.username} already exists")
        return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST, content=error_message.model_dump())
    user_dict = user.model_dump()
    user_dict.update({"id": 6})
    return user_dict


@app.get("/users/{username}", status_code=200, response_model=UserOut)
async def get_user(username: str = Path(..., min_length=1)):
    user = users.get(username, None)
    if user:
        return user
    raise UserNotFoundException(username)


@app.exception_handler(UserNotFoundException)
async def user_not_found_exception_handler(request: Request, exc: UserNotFoundException):
    return JSONResponse(status_code=404,
                        content={
                            "error_code": 404,
                            "message": f"user: {exc.username} not found"})


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

这会儿看结果就会有响应模型 400、401 的

同时重复创建用户的话

依赖注入

  • 依赖注入用于对依赖项的参数进行调用/引用，便于代码的复用，如分页功能

创建一个/depends目录，然后在下面创建main.py

函数作为依赖项

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
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI, Depends

app = FastAPI()


def pageinfo_params(page_index: Optional[int] = 1, page_size: Optional[int] = 10):
    return {"page_index": page_index, "page_size": page_size}


@app.get("/items")
async def get_items(page_info: dict = Depends(pageinfo_params)):
    return page_info


@app.get("/users")
async def get_users(page_info: dict = Depends(pageinfo_params)):
    return page_info


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

类作为依赖项

  • 类作为依赖项时，以下两个表达效果一致
    • page_info: PageInfo = Depends(PageInfo)
    • page_info: PageInfo = Depends()
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
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI, Depends

app = FastAPI()


def pageinfo_params(page_index: Optional[int] = 1, page_size: Optional[int] = 10):
    return {"page_index": page_index, "page_size": page_size}


class PageInfo:
    def __init__(self, page_index: Optional[int] = 1, page_size: Optional[int] = 10):
        self.page_index = page_index
        self.page_size = page_size


@app.get("/items")
async def get_items(page_info: dict = Depends(pageinfo_params)):
    return page_info


@app.get("/users")
async def get_users(page_info: PageInfo = Depends(PageInfo)):
    return page_info


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

子依赖项

子依赖项其实就是 依赖项中某项具有依赖项

增加一个总页码的项

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
34
35
36
37
38
39
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI, Depends

app = FastAPI()


def total_pages_params(total_page: Optional[int] = 1):
    return total_page


def pageinfo_params(page_index: Optional[int] = 1, page_size: Optional[int] = 10,
                    total: Optional[int] = Depends(total_pages_params)):
    return {"page_index": page_index, "page_size": page_size, "total": total}


class PageInfo:
    def __init__(self, page_index: Optional[int] = 1, page_size: Optional[int] = 10,
                 total: Optional[int] = Depends(total_pages_params)):
        self.page_index = page_index
        self.page_size = page_size
        self.total = total


@app.get("/items")
async def get_items(page_info: dict = Depends(pageinfo_params)):
    return page_info


@app.get("/users")
async def get_users(page_info: PageInfo = Depends(PageInfo)):
    return page_info


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

通过依赖注解使用 dependencies

给 get_users 加上一个鉴权的功能

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
34
35
36
37
38
39
40
41
42
43
44
45
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI, Depends, Header, HTTPException

app = FastAPI()


async def auth_verify(api_token: Optional[str] = Header(default=None, alias="api-token")):
    if not api_token:
        raise HTTPException(status_code=401, detail="No API token provided,Unauthorized")


def total_pages_params(total_page: Optional[int] = 1):
    return total_page


def pageinfo_params(page_index: Optional[int] = 1, page_size: Optional[int] = 10,
                    total: Optional[int] = Depends(total_pages_params)):
    return {"page_index": page_index, "page_size": page_size, "total": total}


class PageInfo:
    def __init__(self, page_index: Optional[int] = 1, page_size: Optional[int] = 10,
                 total: Optional[int] = Depends(total_pages_params)):
        self.page_index = page_index
        self.page_size = page_size
        self.total = total


@app.get("/items")
async def get_items(page_info: dict = Depends(pageinfo_params)):
    return page_info


@app.get("/users", dependencies=[Depends(auth_verify)])
async def get_users(page_info: PageInfo = Depends(PageInfo)):
    return page_info


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

全局依赖注入

如果不只是 get_user 需要鉴权，所有 api 都需要鉴权，这是可以在构造 app 的时候传入 dependencies

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
34
35
36
37
38
39
40
41
42
43
44
45
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import Optional

import uvicorn
from fastapi import FastAPI, Depends, Header, HTTPException


async def auth_verify(api_token: Optional[str] = Header(default=None, alias="api-token")):
    if not api_token:
        raise HTTPException(status_code=401, detail="No API token provided,Unauthorized")


app = FastAPI(dependencies=[Depends(auth_verify)])


def total_pages_params(total_page: Optional[int] = 1):
    return total_page


def pageinfo_params(page_index: Optional[int] = 1, page_size: Optional[int] = 10,
                    total: Optional[int] = Depends(total_pages_params)):
    return {"page_index": page_index, "page_size": page_size, "total": total}


class PageInfo:
    def __init__(self, page_index: Optional[int] = 1, page_size: Optional[int] = 10,
                 total: Optional[int] = Depends(total_pages_params)):
        self.page_index = page_index
        self.page_size = page_size
        self.total = total


@app.get("/items")
async def get_items(page_info: dict = Depends(pageinfo_params)):
    return page_info


@app.get("/users")
async def get_users(page_info: PageInfo = Depends(PageInfo)):
    return page_info


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

API 的身份认证

插件准备

  • python-multipart==0.0.6
  • PyJWT==2.8.0
1
2
3
fastapi[all]
python-multipart==0.0.6
PyJWT==2.8.0

用户登录的 API

创建一个目录/security然后创建一个main.py

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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime, timezone, timedelta

import jwt
import uvicorn
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

SECURITY_KEY = "zacharysecret"
ALGORITHMS = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")


class Token(BaseModel):
    access_token: str
    token_type: str


app = FastAPI()


def validate_user(username: str, password: str):
    if username == "zachary" and password == "123456":
        return username
    return None


def get_current_user(token: str = Depends(oauth2_scheme)):
    unauth_exp = HTTPException(status_code=401, detail="Unauthorized")
    try:
        username = None
        token_data = jwt.decode(token, SECURITY_KEY, ALGORITHMS)
        if token_data:
            username = token_data.get("username", None)
    except Exception as e:
        raise unauth_exp

    if not username:
        raise unauth_exp

    return username


@app.post("/token")
async def login(login_form: OAuth2PasswordRequestForm = Depends()):
    username = validate_user(login_form.username, login_form.password)
    if not username:
        raise HTTPException(status_code=401, detail="Incorrect username or password",
                            headers={"WWW-Authenticate": "Bearer"})
    token_expires = datetime.now(timezone.utc) + timedelta(minutes=30)
    token_data = {"username": username, "exp": token_expires}

    token = jwt.encode(token_data, SECURITY_KEY, ALGORITHMS)
    return Token(access_token=token, token_type="bearer")


@app.get("/items")
async def get_items(username: str = Depends(get_current_user)):
    return {"current_user": username}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

可以点击小锁头 先进行一个登录

然后再实现需要登录的 api

连接数据库

插件准备

  • mysqlclient
  • SQLAlchemy
1
2
3
4
5
fastapi[all]
python-multipart==0.0.6
PyJWT==2.8.0
mysqlclient==2.1.1
SQLAlchemy==2.0.23

定义数据库表映射类

创建一个/db_process目录，然后创建一个main.py

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
34
# -*- coding:utf-8 -*-
# Author: Zachary
import uvicorn
from fastapi import FastAPI
from sqlalchemy import create_engine, Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker


class Base(DeclarativeBase):
    pass


engine = create_engine("mysql+mysqldb://root:980226@localhost:3306/testApi", echo=True)


# 定义数据库映射类
class StudentEntity(Base):
    __tablename__ = "students"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
    gender: Mapped[str] = mapped_column(String(10), nullable=False)


Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

app = FastAPI()


# 定义API内容


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

如果现在运行一下这个代码，就会生成 students 这张表

运行后

准备 API 的输入输出 schema 类

1
2
3
4
5
6
7
8
9
10
11
12
13
...
# 定义API模型
class StudentBase(BaseModel):
    name: str
    gender: str


class StudentCreate(StudentBase):
    pass


class StudentOut(StudentBase):
    id: int

准备用户数据库连接的 depends 依赖项

1
2
3
4
5
6
def get_bd_session():
    session = Session()
    try:
        yield session
    finally:
        session.close()

创建和查看所有用户 API

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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import List

import uvicorn
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Integer, String, select, asc
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker


class Base(DeclarativeBase):
    pass


engine = create_engine("mysql+mysqldb://root:980226@localhost:3306/testApi", echo=True)


# 定义数据库映射类
class StudentEntity(Base):
    __tablename__ = "students"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
    gender: Mapped[str] = mapped_column(String(10), nullable=False)


Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

app = FastAPI()


# 定义API模型
class StudentBase(BaseModel):
    name: str
    gender: str


class StudentCreate(StudentBase):
    pass


class StudentOut(StudentBase):
    id: int


def get_bd_session():
    session = Session()
    try:
        yield session
    finally:
        session.close()


@app.post('/students', response_model=StudentOut)
async def create_student(student: StudentCreate, db_session: Session = Depends(get_bd_session)):
    query = select(StudentEntity).where(StudentEntity.name == student.name)
    result = db_session.execute(query).scalars().all()
    if result:
        raise HTTPException(status_code=400, detail=f"该学生({student.name})已存在")

    student = StudentEntity(name=student.name, gender=student.gender)
    db_session.add(student)
    db_session.commit()
    return student


@app.get('/students', response_model=List[StudentOut])
async def get_students(db_session: Session = Depends(get_bd_session)):
    query = select(StudentEntity).order_by(asc(StudentEntity.name))
    return db_session.execute(query).scalars().all()


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

创建一个不存在的用户：成功

查看所有学生列表：成功

创建一个已经存在的学生：异常

实现更新与删除

在/db_process/main.py中做修改

更新

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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import List

import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Path
from pydantic import BaseModel
from sqlalchemy import create_engine, Integer, String, select, asc
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker


class Base(DeclarativeBase):
    pass


engine = create_engine("mysql+mysqldb://root:980226@localhost:3306/testApi", echo=True)


# 定义数据库映射类
class StudentEntity(Base):
    __tablename__ = "students"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
    gender: Mapped[str] = mapped_column(String(10), nullable=False)


Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

app = FastAPI()


# 定义API模型
class StudentBase(BaseModel):
    name: str
    gender: str


class StudentCreate(StudentBase):
    pass


class StudentUpdate(StudentBase):
    pass


class StudentOut(StudentBase):
    id: int


def get_bd_session():
    session = Session()
    try:
        yield session
    finally:
        session.close()


@app.post('/students', response_model=StudentOut)
async def create_student(student: StudentCreate, db_session: Session = Depends(get_bd_session)):
    query = select(StudentEntity).where(StudentEntity.name == student.name)
    result = db_session.execute(query).scalars().all()
    if result:
        raise HTTPException(status_code=400, detail=f"该学生({student.name})已存在")

    student = StudentEntity(name=student.name, gender=student.gender)
    db_session.add(student)
    db_session.commit()
    return student


@app.get('/students', response_model=List[StudentOut])
async def get_students(db_session: Session = Depends(get_bd_session)):
    query = select(StudentEntity).order_by(asc(StudentEntity.name))
    return db_session.execute(query).scalars().all()


@app.put('/students/{student_id}', response_model=StudentOut)
async def update_student(*, student_id: int = Path(...), student: StudentUpdate,
                         db_session: Session = Depends(get_bd_session)):
    query = select(StudentEntity).where(StudentEntity.id == student_id)
    exist_student = db_session.execute(query).scalar()
    if not exist_student:
        raise HTTPException(status_code=404, detail=f"该学生({student_id})不存在")
    exist_student.name = student.name
    exist_student.gender = student.gender
    db_session.commit()
    return exist_student


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

更新已经可以实现了，但是这块仍旧会有些问题：

  • 如果 StudentUpdate 有更新的话，就会需要修改字段更新的部分
  • 如果字段比较多的话需要写的内容也比较多
  • 虽然 SQLAlchemy 提供了 update 方法可以直接更新，但是更新的内容不能直接返回，还需要重新查一次数据库才行

这块列出三种更新方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def set_attrs(obj, data: dict):
    if data:
        for key, value in data.items():
            setattr(obj, key, value)

# 更新方法1，缺点：无法直接获取返回的对象，需要再查一次库，优点：很方便
# update_query = update(StudentEntity).where(StudentEntity.id == student_id).values(**student.dict())
update_query = update(StudentEntity).where(StudentEntity.id == student_id).values(student.model_dump())
db_session.execute(update_query)

# 更新方法2，缺点：如果字段很多，需要一个一个写很麻烦，如果实体类改变字段也需要改 优点：可以直接修改内存中的对象，修改后可以返回
exist_student.name = student.name
exist_student.gender = student.gender

# 更新方法3 优点：可以获取内存中的对象，便于返回；同时如果实体类改变，也会对应修改
set_attrs(exist_student, student.dict())

完整代码

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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import List

import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Path
from pydantic import BaseModel
from sqlalchemy import create_engine, Integer, String, select, asc, update
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker


def set_attrs(obj, data: dict):
    if data:
        for key, value in data.items():
            setattr(obj, key, value)


class Base(DeclarativeBase):
    pass


engine = create_engine("mysql+mysqldb://root:980226@localhost:3306/testApi", echo=True)


# 定义数据库映射类
class StudentEntity(Base):
    __tablename__ = "students"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
    gender: Mapped[str] = mapped_column(String(10), nullable=False)


Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

app = FastAPI()


# 定义API模型
class StudentBase(BaseModel):
    name: str
    gender: str


class StudentCreate(StudentBase):
    pass


class StudentUpdate(StudentBase):
    pass


class StudentOut(StudentBase):
    id: int


def get_bd_session():
    session = Session()
    try:
        yield session
    finally:
        session.close()


@app.post('/students', response_model=StudentOut)
async def create_student(student: StudentCreate, db_session: Session = Depends(get_bd_session)):
    query = select(StudentEntity).where(StudentEntity.name == student.name)
    result = db_session.execute(query).scalars().all()
    if result:
        raise HTTPException(status_code=400, detail=f"该学生({student.name})已存在")

    student = StudentEntity(name=student.name, gender=student.gender)
    db_session.add(student)
    db_session.commit()
    return student


@app.get('/students', response_model=List[StudentOut])
async def get_students(db_session: Session = Depends(get_bd_session)):
    query = select(StudentEntity).order_by(asc(StudentEntity.name))
    return db_session.execute(query).scalars().all()


@app.put('/students/{student_id}', response_model=StudentOut)
async def update_student(*, student_id: int = Path(...), student: StudentUpdate,
                         db_session: Session = Depends(get_bd_session)):
    query = select(StudentEntity).where(StudentEntity.id == student_id)
    exist_student = db_session.execute(query).scalar()
    if not exist_student:
        raise HTTPException(status_code=404, detail=f"该学生({student_id})不存在")

    # # 更新方法1，缺点：无法直接获取返回的对象，需要再查一次库，优点：很方便
    # # update_query = update(StudentEntity).where(StudentEntity.id == student_id).values(**student.dict())
    # update_query = update(StudentEntity).where(StudentEntity.id == student_id).values(student.model_dump())
    # db_session.execute(update_query)
    #
    # # 更新方法2，缺点：如果字段很多，需要一个一个写很麻烦，如果实体类改变字段也需要改 优点：可以直接修改内存中的对象，修改后可以返回
    # exist_student.name = student.name
    # exist_student.gender = student.gender

    # 更新方法3 优点：可以获取内存中的对象，便于返回；同时如果实体类改变，也会对应修改
    set_attrs(exist_student, student.dict())

    db_session.commit()
    return exist_student


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

删除

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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
# -*- coding:utf-8 -*-
# Author: Zachary
from typing import List

import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Path
from pydantic import BaseModel
from sqlalchemy import create_engine, Integer, String, select, asc, update
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker


def set_attrs(obj, data: dict):
    if data:
        for key, value in data.items():
            setattr(obj, key, value)


class Base(DeclarativeBase):
    pass


engine = create_engine("mysql+mysqldb://root:980226@localhost:3306/testApi", echo=True)


# 定义数据库映射类
class StudentEntity(Base):
    __tablename__ = "students"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
    gender: Mapped[str] = mapped_column(String(10), nullable=False)


Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

app = FastAPI()


# 定义API模型
class StudentBase(BaseModel):
    name: str
    gender: str


class StudentCreate(StudentBase):
    pass


class StudentUpdate(StudentBase):
    pass


class StudentOut(StudentBase):
    id: int


def get_bd_session():
    session = Session()
    try:
        yield session
    finally:
        session.close()


@app.post('/students', response_model=StudentOut)
async def create_student(student: StudentCreate, db_session: Session = Depends(get_bd_session)):
    query = select(StudentEntity).where(StudentEntity.name == student.name)
    result = db_session.execute(query).scalars().all()
    if result:
        raise HTTPException(status_code=400, detail=f"该学生({student.name})已存在")

    student = StudentEntity(name=student.name, gender=student.gender)
    db_session.add(student)
    db_session.commit()
    return student


@app.get('/students', response_model=List[StudentOut])
async def get_students(db_session: Session = Depends(get_bd_session)):
    query = select(StudentEntity).order_by(asc(StudentEntity.name))
    return db_session.execute(query).scalars().all()


def check_student_exist(student_id: int, db_session: Session) -> StudentEntity:
    query = select(StudentEntity).where(StudentEntity.id == student_id)
    exist_student = db_session.execute(query).scalar()
    if not exist_student:
        raise HTTPException(status_code=404, detail=f"该学生({student_id})不存在")
    return exist_student


@app.put('/students/{student_id}', response_model=StudentOut)
async def update_student(*, student_id: int = Path(...), student: StudentUpdate,
                         db_session: Session = Depends(get_bd_session)):
    exist_student = check_student_exist(student_id, db_session)

    # # 更新方法1，缺点：无法直接获取返回的对象，需要再查一次库，优点：很方便
    # # update_query = update(StudentEntity).where(StudentEntity.id == student_id).values(**student.dict())
    # update_query = update(StudentEntity).where(StudentEntity.id == student_id).values(student.model_dump())
    # db_session.execute(update_query)
    #
    # # 更新方法2，缺点：如果字段很多，需要一个一个写很麻烦，如果实体类改变字段也需要改 优点：可以直接修改内存中的对象，修改后可以返回
    # exist_student.name = student.name
    # exist_student.gender = student.gender

    # 更新方法3 优点：可以获取内存中的对象，便于返回；同时如果实体类改变，也会对应修改
    set_attrs(exist_student, student.dict())

    db_session.commit()
    return exist_student


@app.delete('/students/{student_id}')
async def delete_student(student_id: int = Path(...), db_session: Session = Depends(get_bd_session)):
    exist_student = check_student_exist(student_id, db_session)

    db_session.delete(exist_student)
    db_session.commit()
    return exist_student


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

后台任务

为什么需要后台任务

当一个 API 执行任务的时间较长时，调用者不希望一直等待任务完成

创建一个目录/background_task，然后创建一个main.py

任务函数

  • 任务函数是一个普通的带参数或者无参数的函数
  • 可以是 async 函数或者普通函数
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
# -*- coding:utf-8 -*-
# Author: Zachary
import time

import uvicorn
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()


def send_message(message: str):
    print(f"start sending message: {message}")
    time.sleep(5)
    print(f"finish sending message: {message}")
    return True


@app.post("/notify")
async def send_notification(message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_message, message=message)
    return {"message": f"Sending notification: {message} in background"}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

元数据与文档 URL

标题、描述和版本信息

  • title
  • description
  • version

目前默认的 swagger-UI 的标题和版本信息都是默认的，没有描述

上面这些内容都可以在构造 app 的时候设置

项目目录下创建一个/metadata_api，然后创建一个main.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# -*- coding:utf-8 -*-
# Author: Zachary

import uvicorn
from fastapi import FastAPI

app = FastAPI(title="My API about query",
              description="All this API is some query api use to zachary's blog",
              version="1.0.0")


@app.get("/books")
async def get_books():
    return {"books": ["book1", "book2", "book3"]}


@app.get("/users")
async def get_users():
    return {"users": ["user1", "user2", "user3"]}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

服务条款与协议信息

  • terms_of_service服务条款
  • contact联系信息
    • name姓名
    • url连接
    • email邮箱
  • license_info协议信息
    • name协议名称
    • url协议地址
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
# -*- coding:utf-8 -*-
# Author: Zachary

import uvicorn
from fastapi import FastAPI

app = FastAPI(title="My API about query",
              description="All this API is some query api use to zachary's blog",
              version="1.0.0",
              terms_of_service="https://zachary.com",
              contact={
                  "name": "zachary",
                  "url": "https://zachary.com",
                  "email": "zachary@qq.com"
              },
              license_info={
                  "name": "Apache 2.0",
                  "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
              })


@app.get("/books")
async def get_books():
    return {"books": ["book1", "book2", "book3"]}


@app.get("/users")
async def get_users():
    return {"users": ["user1", "user2", "user3"]}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

标签

  • openapi_tags=tags_metadata
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
# -*- coding:utf-8 -*-
# Author: Zachary

import uvicorn
from fastapi import FastAPI

tags_metadata = [
    {
        "name": "books",
        "description": "All this API for books management",
        "externalDocs": {
            "description": "Books external docs",
            "url": "https://www.baidu.com"
        }
    },
    {
        "name": "users",
        "description": "All this API for users management",
    }
]

app = FastAPI(title="My API about query",
              description="All this API is some query api use to zachary's blog",
              version="1.0.0",
              terms_of_service="https://zachary.com",
              contact={
                  "name": "zachary",
                  "url": "https://zachary.com",
                  "email": "zachary@qq.com"
              },
              license_info={
                  "name": "Apache 2.0",
                  "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
              },
              openapi_tags=tags_metadata)


@app.get("/books", tags=["books", "users"])
async def get_books():
    return {"books": ["book1", "book2", "book3"]}


@app.get("/users", tags=["users"])
async def get_users():
    return {"users": ["user1", "user2", "user3"]}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

openapiURL

  • openapi_url="/api/v1/openapi.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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
# -*- coding:utf-8 -*-
# Author: Zachary

import uvicorn
from fastapi import FastAPI

tags_metadata = [
    {
        "name": "books",
        "description": "All this API for books management",
        "externalDocs": {
            "description": "Books external docs",
            "url": "https://www.baidu.com"
        }
    },
    {
        "name": "users",
        "description": "All this API for users management",
    }
]

app = FastAPI(title="My API about query",
              description="All this API is some query api use to zachary's blog",
              version="1.0.0",
              terms_of_service="https://zachary.com",
              contact={
                  "name": "zachary",
                  "url": "https://zachary.com",
                  "email": "zachary@qq.com"
              },
              license_info={
                  "name": "Apache 2.0",
                  "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
              },
              openapi_tags=tags_metadata,
              openapi_url="/api/v1/openapi.json")


@app.get("/books", tags=["books", "users"])
async def get_books():
    return {"books": ["book1", "book2", "book3"]}


@app.get("/users", tags=["users"])
async def get_users():
    return {"users": ["user1", "user2", "user3"]}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

文档 URL

  • docs_url="/xxx"
  • redoc_url="/xxx"
  • 如果设置为 None 为禁用
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
# -*- coding:utf-8 -*-
# Author: Zachary

import uvicorn
from fastapi import FastAPI

tags_metadata = [
    {
        "name": "books",
        "description": "All this API for books management",
        "externalDocs": {
            "description": "Books external docs",
            "url": "https://www.baidu.com"
        }
    },
    {
        "name": "users",
        "description": "All this API for users management",
    }
]

app = FastAPI(title="My API about query",
              description="All this API is some query api use to zachary's blog",
              version="1.0.0",
              terms_of_service="https://zachary.com",
              contact={
                  "name": "zachary",
                  "url": "https://zachary.com",
                  "email": "zachary@qq.com"
              },
              license_info={
                  "name": "Apache 2.0",
                  "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
              },
              openapi_tags=tags_metadata,
              openapi_url="/api/v1/openapi.json",
              redoc_url="/r",
              docs_url="/swagger")


@app.get("/books", tags=["books", "users"])
async def get_books():
    return {"books": ["book1", "book2", "book3"]}


@app.get("/users", tags=["users"])
async def get_users():
    return {"users": ["user1", "user2", "user3"]}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

更新: 2024-02-09 05:17:11
原文: https://www.yuque.com/zacharyblock/cx2om6/ifnigya72ztak2tr