全栈开发

FastAPI官方文档学习03

Posted by

1、错误处理

使用 HTTPException、添加自定义响应头

from fastapi import FastAPI, HTTPException
import fastapi_cdn_host

app = FastAPI()
fastapi_cdn_host.patch_docs(app)

items = {"foo": "The Foo"}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, 
                            detail="Item not found",
                            headers={"X-Error": "There goes my error"})
    return {"item": items[item_id]}

安装自定义异常处理器

import fastapi_cdn_host

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()
fastapi_cdn_host.patch_docs(app)


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name


@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."}
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

2、路径配置操作

路径操作装饰器支持多种配置参数。

status_code状态码

import fastapi_cdn_host
from pydantic import BaseModel
from fastapi import FastAPI, status, Body
from typing import Annotated

app = FastAPI()
fastapi_cdn_host.patch_docs(app)


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


@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: Annotated[Item, Body()]):
    return item

tags参数

  • tags 参数的值是由 str 组成的 list （一般只有一个 str ）
  • tags 用于为路径操作添加标签：

在docs页面分组

@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 这两个参数

@app.get("/items/", tags=["items"], summary="This is summary"
         # ,description="This is description"
        )
async def read_items():
    """
    This is docs 和 description 不能同时使用
    """
    return [{"name": "Foo", "price": 42}]
  • FastAPI 支持从函数文档字符串中读取描述内容。
  • response_description 参数用于定义响应的描述
  • deprecated 参数可以把路径操作标记为弃用

3、json兼容编码器

from fastapi.encoders import jsonable_encoder

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)
    print(json_compatible_item_data)
    fake_db[id] = json_compatible_item_data

4、请求体 - 更新数据

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


@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

5、依赖项

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

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

创建依赖项

依赖项就是一个函数，且可以使用与路径操作函数相同的参数：

# 依赖项函数
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}
# 路径操作函数
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
# 路径操作函数
@app.get("/users/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
  • 依赖项函数的形式和结构与路径操作函数一样。
  • 可以把依赖项当作没有「装饰器」的路径操作函数。
  • 依赖项可以返回各种内容。

本例中的依赖项预期接收如下参数：

  • 类型为 str 的可选查询参数 q
  • 类型为 int 的可选查询参数 skip，默认值是 0
  • 类型为 int 的可选查询参数 limit，默认值是 100

然后，依赖项函数返回包含这些值的 dict。

导入 Depends

from fastapi import Depends

声明依赖项

与在路径操作函数参数中使用 Body、Query 的方式相同，声明依赖项需要使用 Depends 和一个新的参数

app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
  • 只能传给 Depends 一个参数
  • 且该参数必须是可调用对象，比如函数
  • 该函数接收的参数和路径操作函数的参数一样。

接收到新的请求时FastAPI执行以下操作：

  • 用正确的参数调用依赖函数（可依赖项）
  • 获取函数返回的结果
  • 把函数返回的结果赋值给路径函数的参数

这样，只编写一次代码，FastAPI 就可以为多个路径操作共享这段代码 。

要不要使用 async

在普通的 def 路径操作函数中，可以声明异步的 async def 依赖项；也可以在异步的 async def 路径操作函数中声明普通的 def 依赖项。

上述这些操作都是可行的，FastAPI 知道该怎么处理。

6、类作为依赖项

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

FastAPI 调用 CommonQueryParams 类。这将创建该类的一个 "实例"，该实例将作为参数 commons 被传递给你的函数。

# commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)] 可简写为
commons: Annotated[CommonQueryParams, Depends()]

7、子依赖项

FastAPI 支持创建含子依赖项的依赖项。

并且，可以按需声明任意深度的子依赖项嵌套层级。

FastAPI 负责处理解析不同深度的子依赖项。

from typing import Annotated

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


def query_extractor(q: str | None = None):
    return q

# 依赖 query_extractor
def query_or_cookie_extractor(
    q: Annotated[str, Depends(query_extractor)],
    last_query: Annotated[str | None, Cookie()] = None,
):
    if not q:
        return last_query
    return q

# 依赖 query_or_cookie_extractor
@app.get("/items/")
async def read_query(
    query_or_default: Annotated[str, Depends(query_or_cookie_extractor)],
):
    return {"q_or_cookie": query_or_default}
  • query_or_cookie_extractor该函数自身是依赖项，但是还依赖另外的依赖项query_extractor
    • 该函数依赖query_extractor，并把query_extractor的返回值赋值给参数q

注意，这里在路径操作函数中只声明了一个依赖项，即 query_or_cookie_extractor 。

但 FastAPI 必须先处理 query_extractor，以便在调用 query_or_cookie_extractor 时使用 query_extractor 返回的结果。

多次使用同一个依赖项

多个依赖项共用一个子依赖项，FastAPI 在处理同一请求时，只调用一次该子依赖项。

FastAPI 不会为同一个请求多次调用同一个依赖项，而是把依赖项的返回值进行「缓存」，并把它传递给同一请求中所有需要使用该返回值的「依赖项」。

如果不想使用「缓存」值：Depends 的参数 use_cache 的值设置为 False

8、路径操作装饰器依赖项

在路径操作装饰器中添加 dependencies 参数

有时，我们并不需要在路径操作函数中使用依赖项的返回值。

或者说，有些依赖项不返回值。

但仍要执行或解析该依赖项。

from fastapi import FastAPI, Header, Depends, HTTPException
from typing import Annotated

app = FastAPI()


async def verify_token(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: Annotated[str, Header()]):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


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

dependencies定义依赖的顺序

触发异常

路径装饰器依赖项与正常的依赖项一样，可以 raise 异常：

async def verify_token(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

返回值

无论路径装饰器依赖项是否返回值，路径操作都不会使用这些值。

9、全局依赖项

为整个应用添加依赖项。

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

10、 使用yield的依赖项

FastAPI支持在完成后执行一些额外步骤的依赖项.

确保在每个依赖中只使用一次 yield。

使用 yield 的数据库依赖项

例如，你可以使用这种方式创建一个数据库会话，并在完成后关闭它。

在发送响应之前，只会执行 yield 语句及之前的代码：

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

yield 语句后面的代码会在创建响应后，发送响应前执行

使用 yield 的子依赖项

from typing import Annotated

from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

下一篇
FastAPI官方文档学习04
上一篇 业务域划分五步实操法