FastAPI Web框架 [1.11]

Posted 2022-03-31 Ch4536251

路径操作配置

#   路径操作配置
#   路径操作装饰器支持多种配置参数。
#   以下参数应直接传递给路径操作装饰器，不能传递给路径操作函数.
#   status_code 状态码:status_code 用于定义路径操作响应中的 HTTP 状态码
#   可以直接传递 int 代码， 比如 404
#   如果记不住数字码的涵义，也可以用 status 的快捷常量
from typing import Optional,Set
from fastapi import  FastAPI,status
from pydantic import BaseModel
app = FastAPI()

class Item(BaseModel):
    name:str
    description:Optional[str] = None
    price:float
    tax:Optional[float] = None
    tags:Set[str] = set()

@app.post("/items/",response_model=Item,status_code=status.HTTP_201_CREATED)
async def create_item(item:Item):
    return item
#   状态码在响应中使用，并会被添加到 OpenAPI 概图
#   也可以使用 from starlette import status 导入状态码。
#   FastAPI 的fastapi.status 和 starlette.status 一样，只是快捷方式。实际上，fastapi.status 直接继承自 Starlette。



#   tags 参数
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = set()      # tags 参数的值是由 str 组成的 list （一般只有一个 str ），tags 用于为路径操作添加标签：

@app.post("/items/",response_model=Item,tags=["items"])
async def create_item(item:Item):
    return item

@app.get("/items/",tags=["items"])
async def read_items():
    return ["name":"Foo","price":"42"]

@app.get("/users/",tags=["users"])
async def read_users():
    return ["username":"johndoe"]



#   summary 和 description 参数
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = set()

@app.post(
    "/items/",
    response_model=Item,
    summary="Create an item",
    description="Create an item with all the information, name, description, price, tax and a set of unique tags",
)
async def create_item(item: Item):
    return item




#   文档字符串（docstring）
#   描述内容比较长且占用多行时，可以在函数的 docstring 中声明路径操作的描述，FastAPI 支持从文档字符串中读取描述内容。
#   文档字符串支持 Markdown，能正确解析和显示 Markdown 的内容，但要注意文档字符串的缩进。
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = set()

@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item



#   响应描述
#   response_description 参数用于定义响应的描述说明:
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = set()

@app.post(
    "/items/",
    response_model=Item,
    summary="Create an item",
    response_description="The created item",
)
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item
#   注意，response_description 只用于描述响应，description 一般则用于描述路径操作。



#   弃用路径操作
#   deprecated 参数可以把路径操作标记为弃用，无需直接删除：
@app.get("/items/", tags=["items"])
async def read_items():
    return ["name": "Foo", "price": 42]

@app.get("/users/", tags=["users"])
async def read_users():
    return ["username": "johndoe"]

@app.get("/elements/", tags=["items"], deprecated=True)
async def read_elements():
    return ["item_id": "Foo"]
#   通过传递参数给路径操作装饰器 ，即可轻松地配置路径操作、添加元数据。

JSON兼容编码器

在某些情况下，您可能需要将数据类型（如 Pydantic 模型）转换为与 JSON 兼容的类型（如dict、list等）。例如，如果您需要将其存储在数据库中。
为此，FastAPI提供了一个 jsonable_encoder() 功能。

#   使用jsonable_encoder
#   假设您有一个 fake_db 只接收 JSON 兼容数据的数据库, 例如，它不接收datetime对象，因为它们与 JSON 不兼容。
#   因此，datetime必须将对象转换为str包含ISO格式数据的对象,同样，该数据库不会接收 Pydantic 模型（具有属性的对象），只会接收dict.你可以使用jsonable_encoder它。
#   它接收一个对象，如 Pydantic 模型，并返回一个 JSON 兼容版本：
#   python3.10以上
from datetime import datetime
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder       #导入 jsonable_encoder
from pydantic import BaseModel
app = FastAPI()
fake_db = 

class Item(BaseModel):
    title: str
    timestamp: datetime
    description: str | None = None

@app.put("/items/id")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    fake_db[id] = json_compatible_item_data
#   在此示例中，它将 Pydantic 模型转换为dict，并将datetime转换为str。
#   调用它的结果是可以用 Python 标准编码的东西json.dumps()
#   它不会返回str包含 JSON 格式数据（作为字符串）的大数据。它返回一个 Python 标准数据结构（例如 dict），其中的值和子值都与 JSON 兼容。
#   jsonable_encoder实际上是FastAPI内部用来转换数据的。但它在许多其他场景中很有用。

请求体 - 更新数据

用 PUT 更新数据
更新数据请用 HTTP PUT 操作：把输入数据转换为以 JSON 格式存储的数据（比如，使用 NoSQL 数据库时），可以使用 jsonable_encoder。例如，把 datetime 转换为 str。

from typing import List, Optional
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
app = FastAPI()

class Item(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: float = 10.5
    tags: List[str] = []

items = 
    "foo": "name": "Foo", "price": 50.2,
    "bar": "name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2,
    "baz": "name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": [],


@app.get("/items/item_id", response_model=Item)
async def read_item(item_id: str):
    return items[item_id]

#   用 PUT 更新数据
@app.put("/items/item_id", response_model=Item)
async def update_item(item_id: str, item: Item):
    update_item_encoded = jsonable_encoder(item)
    items[item_id] = update_item_encoded
    return update_item_encoded
#   PUT 用于接收替换现有数据的数据。
#   关于更新数据的警告
#   用 PUT 把数据项 bar 更新为以下内容时：
#   
#     "name": "Barz",
#     "price": 3,
#     "description": None,
# 
#   因为上述数据未包含已存储的属性 "tax": 20.2，新的输入模型会把 "tax": 10.5 作为默认值。
# 因此，本次操作把 tax 的值「更新」为 10.5。




#   用 PATCH 进行部分更新
#   HTTP PATCH 操作用于更新部分数据。
#   即，只发送要更新的数据，其余数据保持不变.
#   PATCH 没有 PUT 知名，也怎么不常用。很多人甚至只用 PUT 实现部分更新。FastAPI 对此没有任何限制，可以随意互换使用这两种操作。
#
#   使用 Pydantic 的 exclude_unset 参数
#   更新部分数据时，可以在 Pydantic 模型的 .dict() 中使用 exclude_unset 参数。
#   比如，item.dict(exclude_unset=True);这段代码生成的 dict 只包含创建 item 模型时显式设置的数据，而不包括默认值。
#   然后再用它生成一个只含已设置（在请求中所发送）数据，且省略了默认值的 dict：
class Item(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: float = 10.5
    tags: List[str] = []

items = 
    "foo": "name": "Foo", "price": 50.2,
    "bar": "name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2,
    "baz": "name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": [],


@app.get("/items/item_id", response_model=Item)
async def read_item(item_id: str):
    return items[item_id]

@app.patch("/items/item_id", response_model=Item)
async def update_item(item_id: str, item: Item):
    stored_item_data = items[item_id]
    stored_item_model = Item(**stored_item_data)
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item_model.copy(update=update_data)
    items[item_id] = jsonable_encoder(updated_item)
    return updated_item



#   使用 Pydantic 的 update 参数
#   用 .copy() 为已有模型创建调用 update 参数的副本，该参数为包含更新数据的 dict。
#   例如，stored_item_model.copy(update=update_data)：
#   ..
#   ..
@app.patch("/items/item_id", response_model=Item)
async def update_item(item_id: str, item: Item):
    stored_item_data = items[item_id]
    stored_item_model = Item(**stored_item_data)
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item_model.copy(update=update_data)       # 用 .copy() 为已有模型创建调用 update 参数的副本，该参数为包含更新数据的 dict。
    items[item_id] = jsonable_encoder(updated_item)
    return updated_item
#   更新部分数据小结
#   简而言之，更新部分数据应：

#   1 使用 PATCH 而不是 PUT （可选，也可以用 PUT）；
#   2 提取存储的数据；
#   3 把数据放入 Pydantic 模型；
#   4 生成不含输入模型默认值的 dict （使用 exclude_unset 参数）；只更新用户设置过的值，不用模型中的默认值覆盖已存储过的值。
#   5 为已存储的模型创建副本，用接收的数据更新其属性 （使用 update 参数）。
#   6 把模型副本转换为可存入数据库的形式（比如，使用 jsonable_encoder）。
#   7 这种方式与 Pydantic 模型的 .dict() 方法类似，但能确保把值转换为适配 JSON 的数据类型，例如， 把 datetime 转换为 str 。
#   8 把数据保存至数据库；
#   9 返回更新后的模型。
@app.patch("/items/item_id", response_model=Item)
async def update_item(item_id: str, item: Item):
    stored_item_data = items[item_id]
    stored_item_model = Item(**stored_item_data)
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item_model.copy(update=update_data)
    items[item_id] = jsonable_encoder(updated_item)
    return updated_item
#   实际上，HTTP PUT 也可以完成相同的操作。 但本节以 PATCH 为例的原因是，该操作就是为了这种用例创建的。
#   注意，输入模型仍需验证，因此，如果希望接收的部分更新数据可以省略其他所有属性，则要把模型中所有的属性标记为可选（使用默认值或 None）。