全栈开发

FastAPI官方文档学习01

Posted by

运行fastapi程序：

fastapi dev xxx.py

解决fastapi docs页面打不开

# 安装包
pip install fastapi-cdn-host

# 代码示例
from fastapi import FastAPI
import fastapi_cdn_host

app = FastAPI()
fastapi_cdn_host.patch_docs(app)

1、第一步

from fastapi import FastAPI

app = FastAPI()

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

2、路径参数

from fastapi import FastAPI

app = FastAPI()

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

添加参数类型注解

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}
  • 参数限制int类型，其他类型将报错

路由定义的顺序很重要

@app.get("/users/me")
async def read_user_me():
    return {"user_id": "the current user"}

@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}
  • 如果想要正确获取/users/me返回数据，那么必须在/users/{user_id}之前定义

创建枚举类型的路径参数

from enum import Enum

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    if model_name is ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning FTW!"}
    
    if model_name.value == "lenet":
        return {"model_name": model_name, "message": "LeCNN all the images"}
    
    return {"model_name": model_name, "message": "Have some residuals"}
  • 也就是路径参数必须是枚举值域才可以

包含路径的路径参数

@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

3、查询参数

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

@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip: skip + limit]
  • 声明的参数不是路径参数时，路径操作函数会把该参数自动解释为查询参数。
  • 也可以设置查询参数的默认值，即函数参数的默认值

可选参数

将查询参数的默认值设置为None

@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}
  • 0、FALSE、False、false 自动转换为false 其他转换为true

多个路径和查询参数

@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: int, q: str | None = None, short: bool = False):
    item = {"item_id": item_id, "owner_id": user_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "This is an amazing item that has a long description"}
        )
    return item

必选参数

  • 为不是路径参数的参数声明默认值，该参数就不是必选了
  • 如果只想把参数设为可选，但又不想制定参数的值，则要把默认值设为None
  • 如果要把查询参数设置为必选，就不要声明默认值
@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str, skip: int = 0, limit: int | None = None):
    item = {"item": item_id, "needy": needy, "skip": skip, "limit": limit}
    return item
  • needy，必选的str类型参数
  • skip，默认值为0的int类型参数
  • limit，可选的int类型参数

4、请求体

请求体是客户端发送给API的数据。响应体是API发送给客户端的数据。

API基本上都会发送响应体，但是客户端不一定发送请求体。

使用pydantic模型声明请求体

from pydantic import BaseModel


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


@app.post("/items/")
async def create_item(item: Item):
    return item
  • 数据模型声明为继承BaseModel的类
    • 包含默认值的模型属性是可选的，否则就是必选的
    • 默认值为None的模型属性也是可选的

仅使用python类型声明，FastAPI就可以

  • 以JSON形式读取请求体
  • （在必要时）把请求体转换为对应的类型
  • 校验数据：
    • 数据无效时返回错误信息，并指出错误数据的确切位置和内容
  • 把接收的数据赋值给参数item
    • 把函数中请求体参数的类型声明为Item，还能获得代码补全等编辑器支持
  • 为模型生成JSON Schema，在项目中所需的位置使用

请求体 + 路径参数

FastAPI能识别与路径参数匹配的函数参数，还能识别从请求体中获取的类型为Pydantic模型的函数参数

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, **item.model_dump()}

请求体+路径参数+查询参数

FastAPI支持同时声明这三种参数，并从正确的位置获取数据

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: str | None = None):
    result = {"item_id": item_id, **item.model_dump()}
    if q:
        result.update({"q": q})
    return result

函数参数按如下规则进行识别：

  • 路径中声明了相同名称的参数，是路径参数
  • 类型是（int、float、str、bool等）单类型的参数，是查询参数
  • 类型是Pydantic模型的参数，是请求体

5、查询参数和字符串校验

额外的校验

导入Query，使用Query为查询参数声明更多的校验和元数据

from typing import Annotated

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

q: Annotated[str | None, Query(max_length=50)] = None

  • str | None：参数 q 的类型可以是字符串或者 None
  • Query(max_length=50)：这是 FastAPI 的元数据，告诉 FastAPI：
    • 这个参数来自 URL 查询参数（query parameter）
    • 对参数值进行验证：最大长度不能超过 50 个字符
  • = None：默认值为 None，表示这个参数是可选的
  • 使用Annotated 类型信息和默认值分离，代码更易读
    • 如果没有校验规则，使用Annotated也要加Query()

添加正则表达式

q: Annotated[str | None, Query(min_length=3, max_length=50, pattern="^fixedquery$")] = None

声明为必须参数

q: Annotated[str | None, Query(min_length=3]

更多参数：

q: Annotated[
        str | None,
        Query(
            alias="item-query",
            title="Query string",
            description="Query string for the items to search in the database that have a good match",
            min_length=3,
            max_length=50,
            pattern="^fixedquery$",
            deprecated=True,
        ),
    ] = None
  • alias="item-query" 将不会使用q 在url中使用item-query
  • deprecated=True 弃用参数

6、路径参数和数值校验

导入Path，使用Path为路径参数声明更多的校验和元数据

from fastapi import FastAPI, Path, Query
import fastapi_cdn_host
from typing import Annotated


app = FastAPI()
fastapi_cdn_host.patch_docs(app)


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

使用*作为函数的第一个参数，后面的所有参数都应作为关键字参数kwargs来调用，即使他们没有默认值。

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

更多校验

  • gt：大于（greater than）
  • ge：大于等于（greater than or equal）
  • lt：小于（less than）
  • le：小于等于（less than or equal）

7、查询参数模型

在一个Pydantic模型中声明你需要的查询参数，然后为参数添加元数据Query()

from typing import Annotated, Literal
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
import fastapi_cdn_host

app = FastAPI()
fastapi_cdn_host.patch_docs(app)


class FilterParams(BaseModel):
    limit: int = Field(100, gt=0, le=100)
    offset: int = Field(0, ge=0)
    order_by: Literal["created_at", "updated_at"] = "created_at"
    tags: list[str] = []


@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
    return filter_query
http://localhost:8000/items/?limit=100&offset=0&order_by=created_at&tags=a&tags=b
  • FastAPI将会从请求的查询参数中提取出每个字段的数据，并将其提供给定义的Pydantic模型
  • Literal 是一种轻量级的、基于字面量的“伪枚举”

禁止额外的查询参数

class FilterParams(BaseModel):
    model_config = {"extra": "forbid"}

    limit: int = Field(100, gt=0, le=100)
    offset: int = Field(0, ge=0)
    order_by: Literal["created_at", "updated_at"] = "created_at"
    tags: list[str] = []
  • model_config = {"extra": "forbid"}

8、请求体-多个参数

混合使用Path、Query和请求体参数

from fastapi import FastAPI, Query, Path
import fastapi_cdn_host
from typing import Annotated
from pydantic import BaseModel

app = FastAPI()
fastapi_cdn_host.patch_docs(app)


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


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

多个请求体参数

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


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

请求体传参：

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    }
}

与但个请求体参数不同的是：变成了参数名对应请求体json嵌套的形式

请求体中的单一值

from fastapi import Body

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

请求体传参

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    },
    "importance": 5
}
  • Body()也支持转换数据类型，校验，生成文档等。

多个请求体参数和查询参数

@app.put("/items/{item_id}")
async def update_item(
    *,
    item_id: int,
    item: Item,
    user: User,
    importance: Annotated[int, Body(gt=0)],
    q: str | None = None,
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    if q:
        results.update({"q": q})
    return results

由于默认情况下，但一值被解释为查询参数，因此不必显示的添加Query

单个请求体转嵌套传参

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
    results = {"item_id": item_id, "item": item}
    return results
  • 请求体使用Annotated 添加元数据Body() 显示设置嵌套embed=True

9、请求体-字段

使用Pydantic的Field在模型内部声明和校验元数据

class Item(BaseModel):
    name: str
    description: str | None = Field(
        default=None,
        title="item desc",
        max_length=300
    )
    price: float = Field(
        gt=0,
        description="The price must be >= 0"
    )
    tax: float | None = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
    results = {"item_id": item_id, "item": item}
    return results

10、请求体-嵌套模型

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


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


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results
  • 将模型用作一个属性的类型
{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2,
    "tags": ["rock", "metal", "bar"],
    "image": {
        "url": "http://example.com/baz.jpg",
        "name": "The Foo live"
    }
}

带有一组子模型的属性

class Item(BaseModel):
    ...
    images: list[Image] | None = None

任意dict构成的请求体

@app.post("/index-weights/")
async def create_index_weights(weights: dict[int, float]):
    return weights
  • 将接受任意键为 int 类型并且值为 float 类型的 dict

doc接口文档添加示例

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

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "name": "Foo",
                    "description": "A very nice Item",
                    "price": 35.4,
                    "tax": 3.2,
                }
            ]
        }
    }

单独指定字段示例

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

请求体Body也可以指定示例

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Annotated[
        Item,
        Body(
            examples=[
                {
                    "name": "Foo",
                    "description": "A very nice Item",
                    "price": 35.4,
                    "tax": 3.2,
                }
            ],
        ),
    ],
):
    results = {"item_id": item_id, "item": item}
    return results
下一篇
大数据数仓建模体系
上一篇 ShadCN学习记录