【python】FastAPI请求头体（二）

需求场景：假设现在我们有个图书网站，需要客户端在请求头上携带有效的token，才可以获取图书数据，那FastAPI该如何获取客户端传过来的请求头参数呢？

示例1：FastAPI获取，指定请求头的值（比如，请求头的key 是token）

from fastapi import FastAPI, Header # 引入Header类

app = FastAPI()
books = {i: f"图书{i}" for i in range(10)}


@app.get("/books/{id}")
def get_book_by_id(id: int, token: str = Header()):# token=Header() token是Header 的实例化对象
    if token == "this-is-token":
        return books.get(id)
    else:
        return "invalid token"

1、Header和Path() Query()类似，也具有基本的请求头参数校验和API文档设置

2、如果定义的参数是请求头参数，则必须使用Header() 否则FastAPI默认会把这个参数识别的查询参数

3-2 使用请求体提交数据【重点】

向服务器提交数据，可以通过之前咱们学的路径参数、查询参数、也可以在请求体中携带数据（推荐用法）

推荐使用请求体向服务器提交数据的原因：

路径参数和查询参数携带的数据量有限
路径参数和查询参数携带数据不安全

那我们该如何使用请求体向服务端提交数据呢？一般有两种方式：【二选一】

使用POST请求，在请求体中使用JSON格式的数据；PUT/PATCH等也可以在请求体中使用JSON格式数据
使用POST请求，在请求体中使用form-data格式的数据，提交数据（即前端的form表单，第6章详细介绍）

实现方式：使用Pydantic 来接收请求体数据

第一步：继承Pydantic的BaseModel,定义模型类User,再类钟定义两个和请求体字段同名的类数学并做类型提示

第二步：再路径函数钟定义形参user ,设置类型未User

第三步：通过对象user属性username和password获取请求体字段的值

from fastapi import FastAPI
from pydantic import BaseModel # 引用基础模型
app  = FastAPI()

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


@app.post("/login")
def login(user: User):
    return {
        "username": user.username,
        "password": user.password
    }

这样可以帮我们获取到请求体参数并赋值到参数user，同时：

1、按照json格式来读取请求体数据
2、把类型转换且进行类型校验（类型必须符合User格式）
3、把请求体中每个字段的值赋值到模型类对象的每个属性（赋值到User类，基础自基础模型类）
4、API文档可以看到类型提示

结论：

如果请求体参数是JSON格式，就通过继承Pydantic的 BaseModel 的方式来获取数据
BaseModel的子类中定义的类属性是用来匹配请求体对应的字段值，所以名字包保持一致，字段的顺序没有要求
只要路径函数式中形参的类型是BaseModel的子类，则FastAPI会默认会把请求体数据保存在这个变量中
定义的模型类的属性，可以有默认值

3-4 多请求体入参不常用

原理是一样的，就是多定义一个模型而已，不常用拉

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


class Item(BaseModel):
    name: str
    description: typing.Optional[str] = None  # 选填类型，Optional可选的意思
    price: float
    tax: typing.Optional[float] = None  # 选填类型，Optional可选的意思
@app.post("/login_double")
def post_login_double(user: User, item: Item):
    return {"user": {
            "username": user.username,
            "password": user.password,
        },"item": {
            "name": item.name,
            "description": item.description,
            "price": item.price,
            "tax": item.tax,}}

3-5 使用Body接收请求体数据

当你需要和其他入参一起发出请求的时候，比如说

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "liuxu",
        "password": "liuxu"
    },
    "importance": 5
}

上面的两种情况，直接使用继承Pydantic的BaseModel定义模型类的方式行不通，需要借助Body()函数来帮忙

示例1：使用Body()获取请求体中单一字段变量，比如需求1中的 "importance": 5

class User(BaseModel):
    username: int
    password: str


class Item(BaseModel):
    name: str
    description: typing.Optional[str] = None
    price: float
    tax: typing.Optional[float] = None


@app.post("/login")
def login(user: User, item: Item, importance: int = Body()):	# 定义和请求体字段同名的形参并使用Body()
    return importance

1、 Body()函数的室友和前面的Path() Query(),Header()基本一致

2、如果你仅仅需要从请求体中获取importance，其他字段不需要，可直接：def login(importance: int = Body()):

3.6 请求体使用Body校验

Body和前面接收的 Path、Query、Header一样，都可以做数值校验和字符串的基本校验
只不过Body校验的是请求体中的单一数据，比如请求体中只有一个数字或者一个字符串才可以
除此之外，还可以设置请求体的类型，默认是application/json，即JSON格式

from fastapi import FastAPI, Body

app = FastAPI()


@app.post("/login")
def login(name: str = Body(min_length=3), age: int = Body(ge=18)):
    return {"name": name, "age": age}

3.7 模型类嵌套

客户端传过来的请求体数据结构是复杂的嵌套结构，比如下面这种，此时后端该如何接收数据呢？

需求3：嵌套列表中的模型类

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2,
    "images": [
        {
            "url": "http://example.com/baz.jpg",
            "name": "The Foo live"
        },
        {
            "url": "http://example.com/dave.jpg",
            "name": "The Baz"
        }
    ]
}

示例3：解决嵌套列表中的模型类

from typing import List, Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Image(BaseModel):
    url: str
    name: str


class Item(BaseModel):
    name: str
    price: float
    images: Optional[List[Image]] = None


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

3.8 请求体字段使用Filed校验

请求体的每个字段可以单独做校验吗，比如：

name的长度最少是3，
price不少于10
`tags字段限制元素不重复，限制元素的个数等等

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2,
    "tags": ["rock", "metal", "bar"]
}

示例1：使用 Field来校验每个请求体字段，注意，Field不是从FastAPI导入的，而是直接从Pydantic导入的

from typing import Optional, List

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class Item(BaseModel):
    name: str = Field(min_length=5)
    description: Optional[str] = Field(default=None, max_length=300)
    price: float = Field(gt=0, multiple_of=2)
    tax: List[str] = Field(min_items=2, max_items=5, unique_items=True)


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

校验参数总结：

gt/ge/lt/le:				# 数值大小校验(float)
min_length/max_length	    # 字符串长度校验(int)
min_items/max_items			# 集合元素数量校验（int）
unique_items				# 集合元素是否可以重复(布尔值)
multiple_of					# 限制值的某个数字的倍数（float）

max_digits					# 最长多少位数字【bug】
decimal_places				#　最多几位小数【bug】

3.9 总结获取请求体数据的3种方式

使用Pydantic【主要】

定义模型类BaseModle，和请求体中的字段保持一致，定义模型类的属性并做类型提示（支持类型校验和类型转换）

使用Body()

有个特殊的请求体格式，使用Body()显示获取，比如，单一字段的值，内嵌字段的值
可以与其他参数一起使用：Body() 函数可以与其他参数（如路径参数、查询参数等）一起使用，以便更灵活地定义请求处理函数的参数。
支持更多的参数配置：Body() 函数支持传递更多参数配置，如数据类型、默认值等。

使用Python的字典类型和列表类型

定义在路径函数内的特殊字段会默认被视为请求体参数，比如列表，集合，元祖，字典（容器型数据）

1
2
3

@app.post("/item")
def create_item(item: typing.Dict[str, str]): # 直接在路径上
    return item

也可以通过@dataclasses.dataclass类装饰器转装饰成数据类

@dataclasses.dataclass
class User:
    username: str
    password: str


@app.post("/item")
def create_item(user: User):
    return user

3.10 给请求体提供范例

提供范例后，再阅读API文档的时候更方便

FastAPI提供了多种为请求体提供范例的方式

示例1：在模型类中定义 Config类

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None

    class Config:
        schema_extra = {
            "example": {
                "name": "Foo",
                "description": "A very nice Item",
                "price": 35.4,
                "tax": 3.2,
            }
        }


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

示例2：使用 Field(example="xxx")为每个字段提供例子

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class Item(BaseModel):
    name: str = Field(example="Foo")
    description: Union[str, None] = Field(default=None, example="A very nice Item")
    price: float = Field(example=35.4)
    tax: Union[float, None] = Field(default=None, example=3.2)


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

示例3：使用 Body(example="一个例子", examples="多组例子")，这种方式也适用：Path\Query\Header等等【不常用】

from typing import Union

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None


@app.post("/item")
def create_item(
    *,
    item: Item = Body(
        examples={
            "normal": {
                "summary": "A normal example",
                "description": "A **normal** item works correctly.",
                "value": {
                    "name": "Foo",
                    "description": "A very nice Item",
                    "price": 35.4,
                    "tax": 3.2,
                },
            },
            "converted": {
                "summary": "An example with converted data",
                "description": "FastAPI can convert price `strings` to actual `numbers` automatically",
                "value": {
                    "name": "Bar",
                    "price": "35.4",
                },
            },
            "invalid": {
                "summary": "Invalid data is rejected with an error",
                "value": {
                    "name": "Baz",
                    "price": "thirty five point four",
                },
            },
        },
    ),
):
    return item

本章学习总结，

主要是在与对请求头Header()和请求体Body()的定义
这里的Header的定义只是一个简单的定义，还没有注入依赖项（token令牌等）
请求体的话主要使用Pydantic模型类BoseModel定义,在路径函数上做类型提示更方便

4-1 设置响应状态码

示例1：直接给status_code字段一个状态码

@app.post("/login", status_code=200)	# 直接指定一个响应码
def login(user: User):
    return {
        "username": user.username,
        "password": user.password
    }

示例3：status_code使用常量

from fastapi import FastAPI, Path, Header,status
@app.post("/login", status_code=status.HTTP_200_OK)
def login():
    pass

4-2 设置响应头

有的时候，在响应客户端请求的时候，需要在响应头中给客户端返回一些数据，此时该如何设置响应头呢？

示例1：在路径函数中使用 Response

from fastapi import FastAPI, Response

app = FastAPI()


@app.post("/login")
def login(response: Response):
    response.headers["x-jwt-token"] = "this_is_jwt_token"
    return {"message": "Hello World"}

示例2：直接使用返回一个Response，比如使用FastAPI的默认响应方式JSONResponse

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()


@app.post("/login")
def login():
    response = JSONResponse(
        content={"message": "Hello World"},
        headers={"x-jwt-token": "this_is_jwt_token"},
    )
    return response

扩展：使用返回Response的方式，也可以设置响应状态码，手动设置响应体，操作cookie等等

4-3 响应体数据

FastAPI响应数据时，默认响应是json的格式，具体使用的响应类是 JSONResponse

举两个响应示例：一个Pydantic模型类，一个Response对象或 Response的子类

示例3：响应基于Pydantic的模型类

from fastapi import FastAPI
from pydantic import BaseModel
from fastapi.responses import JSONResponse

app = FastAPI(default_response_class=JSONResponse) # 默认响应类JSONResponse


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


@app.post("/login")
def login(user: User):
    return user

示例4：直接响应一个 Response对象或 Response的子类

from fastapi import FastAPI, Response
import json

app = FastAPI()


@app.get("/books")
def get_books():
    response = Response(
        content=json.dumps([{"id": i, "name": f"图书{i}"} for i in range(1, 11)]),
        media_type="application/json"
    )
    return response

4-4 使用响应模型response_model

需求场景

需求1：用户注册接口（需要用户名和密码），后端完成注册后一般会将用户信息返回，但此时不能返回用户密码，该如何操作？
需求2：想让接口返回的字段按照一定的规范来，有数据格式校验和转换等功能，该怎么办？
需求3：API文档展示响应字段和类型

FastAPI提供了响应模型（response_model）的概念，就像定义模型类用来接收请求体数据，响应模型用来处理响应数据。

利用响应模型，可以实现返回字段的动态过滤，类型校验和类型转换等。

class User(BaseModel):
    username: str
    password: str
    email: str


class UserOut(BaseModel):
    username: str
    email: str
@app.post("/registe", response_model=UserOut)		# 指定了响应模型是UserOut，则只返回username和email两个字段 过滤掉了
def registe(user: User):
    return user

示例3：使用response_model做类型校验和类型转换

import typing

from fastapi import FastAPI

app = FastAPI()


@app.post("/demo", response_model=typing.Dict[str, int])
def demo():
    return {"code": 200, "value": "100"}		# 像一个数据的值，会自动转换为Int类型 这里就做了强制的类型转换

4-5 用于过滤响应数据的五个参数

很多时候，我们需要对响应模型中的某些字段做一个过滤操作，比如，

没有赋值的字段过滤掉
使用默认值的字段过滤掉
指定某些字段过滤掉
保留指定的字段等等

FastAPI提供了5个配合响应模型一块使用的字段：

response_model_include，只展示选中的字段
response_model_exclude，选中的字段不展示
response_model_exclude_unset，过滤掉没有赋值的字段
response_model_exclude_defaults，过滤掉使用默认值的字段
response_model_exclude_none，排序没有值的字段

@app.get(
    "/items/{name}",
    response_model=Item,
    response_model_include={"name"}, # 只展示选中的
    response_model_exclude={"tax"}, # 选中的不展示的
    response_model_exclude_unset=True, # 过滤，过滤掉没有赋值的字段
    response_model_exclude_defaults=True, # 过滤，过滤掉使用了默认值得字段
    response_model_exclude_none=True, # 排序，排序没有值的字段
)
def read_item_name(name: str):
    return items[name]

4-6 Pydantic模型类对象的dict方法

这个方法好像弃用了

4-7 jsonable_encoder【可能是重点】

客户端向服务端提交的数据需要永久存储保存数据库，那么此时就需要保证数据都是可序列化的，如何保证待序列化的数据都是可序列化的呢?

from datetime import datetime
from typing import Union
import json
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

fake_db = {}


class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Union[str, None] = None


app = FastAPI()


@app.post("/item")
def create_item(item: Item):
    json_compatible_item_data = jsonable_encoder(item) # 转化json可序列化
    print(type(json_compatible_item_data))
    print(json_compatible_item_data)
    json.dumps(json_compatible_item_data)
    return item

4-8 响应部分总结

如何设置状态码：statuc_code、Response及其子类
如何设置响应头：Response及其子类
如何设置响应体：默认响应类是JSONResponse，也可以手动返回Response及其子类的对象
使用响应模型做字段过滤
关于不同类型的响应（下载文件等）在11章响应介绍

5 、阶段性总结

关于博客的增删改查练习

5个请求方式的使用规范
错误处理
Restful风格api设计
RESTful是Web API接口的设计规范风格，这种风格的理念认为后端开发任务就是提供数据的，对外提供的是数据资源的访问接口，尤其适用于前后端分离的应用模式中。所以在定义接口时，客户端访问的URL路径就表示这种要操作的数据资源。

任何一个框架都可以实现符合restful规范的API接口。

RESTful十条规范：

1 数据的安全保障：url链接一般都采用https协议进行传输

2 接口特征表现，一看就知道是个api接口
- 用api关键字标识接口url：https://www.baidu.com/api
3 多数据版本共存，在url链接中标识数据版本
- https://api.baidu.com/v1
- https://api.baidu.com/v2
!!!4 数据即是资源，均使用名词（可复数）
- 接口一般都是完成前后台数据的交互，交互的数据我们称之为资源
- https://api.baidu.com/users
  注：一般提倡用资源的复数形式，在url链接中奖励不要出现操作资源的动词
- 特殊的接口可以出现动词，因为这些接口一般没有一个明确的资源，或是动词就是接口的核心含义
- https://api.baidu.com/place/search
- https://api.baidu.com/login
!!!5 资源操作由请求方式决定（method）
- 操作资源一般都会涉及到增删改查，使用请求方式确定操作类型
  - https://api.baidu.com/books - get请求：获取所有书
  - https://api.baidu.com/books/1 - get请求：获取主键为1的书
  - https://api.baidu.com/books - post请求：新增一本书书
  - https://api.baidu.com/books/1 - put请求：整体修改主键为1的书
  - https://api.baidu.com/books/1 - patch请求：局部修改主键为1的书
  - https://api.baidu.com/books/1 - delete请求：删除主键为1的书
6 过滤，通过在url上传参的形式传递搜索条件，或者说查询参数(query_params)
7 响应状态码

8 错误处理
9 返回结果，针对不同操作，服务器向用户返回的结果应该符合以下规范
GET /collection：返回资源对象的列表（数组）
GET /collection/resource：返回单个资源对象
POST /collection：返回新生成的资源对象
PUT /collection/resource：返回完整的资源对象
PATCH /collection/resource：返回完整的资源对象
DELETE /collection/resource：返回一个空文档
10 需要url请求的资源需要访问资源的请求链接
- Hypermedia API，RESTful API最好做到Hypermedia，
- 即返回结果中提供链接，连向其他API方法，使得用户不查文档，也知道下一步应该做什么
1
2
3
4
5
6
7
8
9
10
>{ "status": 0, "msg": "ok", "results":[ { "name":"肯德基(罗餐厅)", "img": "https://image.baidu.com/kfc/001.png" } ] >}

源码示例：模拟RESTFul风格的api

import typing

from fastapi import FastAPI, HTTPException, status
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI(title="Blog CRUD")


# mock db
blogs = {
    1: {
        "id": 1,
        "title": "blog1",
        "body": "this is blog1",
        "desc": "desc"
    },
    2: {
        "id": 2,
        "title": "blog2",
        "body": "this is blog2",
        "desc": "desc"
    }
}
# blogs = [
#     {
#         "id": 1,
#         "title": "blog1",
#         "body": "this is blog1",
#         "desc": "desc"
#     },
#     {
#         "id": 2,
#         "title": "blog2",
#         "body": "this is blog2",
#         "desc": "desc"
#     }
# ]


class Blog(BaseModel):
    title: typing.Optional[str] = None
    body: typing.Optional[str] = None
    desc: str


@app.get("/blogs", tags=["Blog"])
def get_blogs(page: int = 1, size: int = 10):
    blogs_list = list(blogs.values())
    return blogs_list[(page - 1) * size:page * size]


@app.get("/blog",  tags=["Blog"])
def get_blog_by_id(blog_id: int):
    blog = blogs.get(blog_id)
    if not blog:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Not found the blog with id: {blog_id}"
        )
    return blog


@app.post("/blog",  tags=["Blog"])
def create_blog(blog: Blog):
    blog_id = len(blogs) + 1
    blogs[blog_id] = {"id": blog_id, **jsonable_encoder(blog)}
    return blogs[blog_id]


@app.put("/blog",  tags=["Blog"])
def update_blog(blog_id: int, blog: Blog):
    to_update_blog = blogs.get(blog_id)
    if not to_update_blog:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Not found the blog with id: {blog_id}"
        )
    to_update_blog.update(jsonable_encoder(blog))
    blogs[blog_id] = to_update_blog
    return to_update_blog


@app.patch("/blog",  tags=["Blog"])
def update_blog2(blog_id: int, blog: Blog):
    to_update_blog = blogs.get(blog_id)
    if not to_update_blog:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Not found the blog with id: {blog_id}"
        )
    to_update_blog.update(**jsonable_encoder(blog, exclude_unset=True))
    blogs[blog_id] = to_update_blog
    return to_update_blog


@app.delete("/blog",  tags=["Blog"])
def delete_blog(blog_id: int):
    if not blogs.get(blog_id):
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Not found the blog with id: {blog_id}"
        )
    return blogs.pop(blog_id, None)

joshua

python

#python #编程，web开发

【python】FastAPI请求头体（二）

http://example.com/2024/01/16/682FastAPI2请求头体/

作者

Wangxiaowang

发布于

2024年1月16日

许可协议

【python】FastAPI表单和文件上传（三）上一篇

【python】FastAPI简单学习和路径参数（一）下一篇

【python】FastAPI请求头体（二）

3-1使用Header 接收请求头参数

3-2 使用请求体提交数据【重点】

3-4 多请求体 入参 不常用

3-5 使用Body接收请求体数据

3.6 请求体使用Body校验

3.7 模型类嵌套

3.8 请求体字段使用Filed校验

3.9 总结获取请求体数据的3种方式

3.10 给请求体提供范例

本章学习总结，

4-1 设置响应状态码

4-2 设置响应头

4-3 响应体数据

4-4 使用响应模型response_model

4-5 用于过滤响应数据的五个参数

4-6 Pydantic模型类对象的dict方法

4-7 jsonable_encoder【可能是重点】

4-8 响应部分总结

5 、阶段性总结

关于博客的增删改查练习

3-4 多请求体入参不常用