Python 之 Fastapi 框架学习

Fastapi 据说有并肩Go的极高性能，我倒是想特别见识一下。

Looooking

1464人浏览 · 2024-04-03 10:43:50

Looooking · 2024-04-03 10:43:50 发布

依赖安装

Fastapi 有版本要求，需要的 Python 版本至少是 Python 3.8（不要犟，按照版本要求来，我最先也是在我 Python3.6 上装的，果不其然跑不起来），幸好我 Win7 老古董能支持的 Python 最高版本可以到 3.8 ，这也算我最后的倔强了。据说这个框架有和 Go 并肩的极高性能，我倒是想特别见识一下。

大家也可以直接去看用户指南教程 - 用户指南 - FastAPI ，这里面讲解得更加详细（中文版的页面偶尔打不开，实在打不开的话建议直接看英文原版——路径中的 /zh 去掉即可）。

安装 fastapi

这个用 pip 安装就是了，没啥多说的，特别是安装有多个 python 解释器的时候，要关注下 pip 和 python 解释器的对应关系。当然，用虚拟环境也是不错的选择。

pip3.8 install fastapi

安装ASGI 服务器

pip3.8 install "uvicorn[standard]"

这点和 flask 不太一样，flask 创建的 app 直接用 app.run() 就跑起来了，它好像不能这么做，需要使用安装的 uvicorn 去启动服务。

Uvicorn是一个基于ASGI（Asynchronous Server Gateway Interface）的异步Web服务器，用于运行异步Python web应用程序。它是由编写FastAPI框架的开发者设计的，旨在提供高性能和低延迟的Web服务。

快速启动

hello world

import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def index():
    """
    注册一个根路径
    :return:
    """
    return {"message": "Hello World"}


if __name__ == "__main__":
    uvicorn.run(app)

文档访问

SwaggerUi风格文档:http://127.0.0.1:8000/docs

有了路由的在线文档，对于接口对接或者 API 接口文档编写的时候，就真正方便多了。而且还可以在页面上直接进行接口请求测试，用起来也很方便。

OpenAPI

FastAPI框架内部实现了OpenAPI 规范，通过访问 http://127.0.0.1:8000/openapi.json,我们可以看到整个项目的 API对应的JSON描述。简单来说，这就是一个结构化的 Swagger 文档，返回的细节很详细。

{
  "openapi": "3.1.0",
  "info": {
    "title": "FastAPI",
    "version": "0.1.0"
  },
  "paths": {
    "/": {
      "get": {
        "summary": "Index",
        "description": "注册一个根路径\n:return:",
        "operationId": "index__get",
        "responses": {
          "200": {
            "description": "Successful Response",
            "content": {
              "application/json": {
                "schema": {}
              }
            }
          }
        }
      }
    }
  }
}

路径参数

参数声明

类似于 Flask 的动态路由，使用 Python 标准类型注解（自从有了类型注解以后，总感觉 Python 变了，不像是原来那个不关心类型的脚步语言了），声明路径操作函数中路径参数的类型。

import uvicorn
from fastapi import FastAPI

app = FastAPI()


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


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

如果输入的参数类型不匹配的话，还会报错。

Please correct the following validation errors and try again.
For 'item_id': Value must be an integer.

路径顺序

import uvicorn
from fastapi import FastAPI

app = FastAPI()


@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}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

比如要使用 /users/me 获取当前用户的数据。然后还要使用 /users/{user_id}，通过用户 ID 获取指定用户的数据。由于路径操作是按顺序依次运行的，因此，一定要在 /users/{user_id} 之前声明 /users/me 。这个其实好理解，相当于先写的动态路由把后面的其他路由通配了。一般如果代码审查比较严格的话，是不允许出现这种情况的。特殊的路由就用特殊的路径，尽量不要和通配路径去碰瓷。

路径预设值

路径操作使用 Python 的 Enum 类型接收预设的路径参数。也就是说和动态路由类似，但是动态值的可选范文是提前限定好的。

import uvicorn
from enum import Enum

from fastapi import FastAPI


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


app = FastAPI()


@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"}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

参数校验

Path 为路径参数声明相同类型的校验和元数据。

import uvicorn

from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
        q: str,
        item_id: int = Path(title="The ID of the item to get", ge=100, le=1000),
        size: float = Query(gt=0, lt=10.5),
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    if size:
        results.update({"size": size})
    return results


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

路径参数大小范围不符

http://127.0.0.1:8000/items/12?q=testing&size=9.9

混合的查询参数范围不符

http://127.0.0.1:8000/items/120?q=testing&size=19.9

正常访问

http://127.0.0.1:8000/items/120?q=testing&size=5.6

查询参数

参数声明

查询参数都是 URL 的组成部分，因此，它们的类型本应是字符串。但声明 Python 类型之后，这些值就会转换为声明的类型，并进行类型校验。

import uvicorn
from fastapi import FastAPI

app = FastAPI()

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]


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

下面的情况，因为没有传入 skip 的值，所以 skip 使用默认值 0 （额外传入的无关参数会被丢弃）

http://127.0.0.1:8000/items/?id=1&limit=1

如果都传入的话，则使用传入的值

http://127.0.0.1:8000/items/?skip=1&limit=1

默认值

即在声明类型的时候，按照上面的示例指定默认值即可。未传入则使用默认值，如果传入的话，则使用传入的值。

可选参数

将参数的默认值设置为 None 之后，就可以将参数变为可选参数（否则就是必选参数）。

import uvicorn
from fastapi import FastAPI

app = FastAPI()

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


@app.get("/items/{item_id}")
async def read_item(item_id: str, query: str = None):
    if query:
        return {"item_id": item_id, "query": query}
    return {"item_id": item_id}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

http://127.0.0.1:8000/items/1?query=2

http://127.0.0.1:8000/items/1

自动类型转换

参数还可以声明为 bool 类型，FastAPI 会自动转换参数类型（常规的 1和0、on和off、yes和no、True和False），其他数据则会报类型转换失败。

import uvicorn
from fastapi import FastAPI

app = FastAPI()


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


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

转换成 True 的示例

http://127.0.0.1:8000/items/1?short=on

http://127.0.0.1:8000/items/1?short=yes

http://127.0.0.1:8000/items/1?short=1

转换成 False 的示例

http://127.0.0.1:8000/items/1?short=0

转换失败的示例

http://127.0.0.1:8000/items/1?short=2

参数校验

Query 显式地将 q 其声明为查询参数。因此查询的时候要以查询参数的格式传参。

import uvicorn

from typing import Union

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(
    q: Union[str, None] = Query(default="fixedquery", min_length=6, max_length=50, pattern="^fixed"),
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

使用默认值

正常访问

长度校验不通过

正则模式不匹配

参数模型

如果有多个参数，可以对一组具有相关性的查询参数，你创建一个 Pydantic 模型来声明它们。

比如以下的参数模型：

禁止了指定查询参数以外的查询参数（正常允许额外的查询参数，只是会被忽略）
指定了查询参数 limit 和 offset 的范围
指定了查询参数 order_by 的值只能从指定的枚举值中选择
指定了查询参数 tags 为字符串列表的形式（一般非列表形式的查询参数，如果像 tags=hello&tags=world 这种指定多个值的话，只会对最后一个赋值生效）

import uvicorn

from typing import Annotated, Literal

from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI()


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] = []


@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
    return filter_query


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

请求体参数

参数声明

这种的提前定义基本上和 golang 就差不多了，请求体必须按照设定的数据模型进行传参，服务端再进行类似 Unmarshal 的操作（这也是速度更快的原因之一），使用指定的结构体接收数据。

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel


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


app = FastAPI()


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


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

这样可以得到以下特征

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

import requests

if __name__ == '__main__':
    req_param = {
        "name": "Looking",
        # "description": "An optional description",
        "price": 2.3,
        "tax": 0.25
    }
    url = r"http://127.0.0.1:8000/items/"
    response = requests.request('POST', url, json=req_param)
    print(response.json())
    # {'name': 'Looking', 'description': None, 'price': 2.3, 'tax': 0.25}

import requests

if __name__ == '__main__':
    req_param = {
        "name": 1234,
        # "description": "An optional description",
        "price": 2.3,
        "tax": 0.25
    }
    url = r"http://127.0.0.1:8000/items/"
    response = requests.request('POST', url, json=req_param)
    print(response.json())
    # {'detail': [{'type': 'string_type', 'loc': ['body', 'name'], 'msg': 'Input should be a valid string', 'input': 1234}]}

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

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

路径中声明了相同参数的参数，是路径参数。如下例中的 item_id，直接拼接在 url 路径中。
类型是（int、float、str、bool 等）单类型参数，是查询参数。如下例中的 importance，请求时需要拼接在 url 的 ? 后边。
类型是 Pydantic 模型的参数，是请求体。如下例中的 item 和 user。

from typing import Annotated

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

app = FastAPI()


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


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


@app.put("/items/{item_id}")
async def update_item(
        item_id: Annotated[int, Path(title="The ID of the item to get", ge=10, le=1000)], item: Item, user: User, importance: int
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例：

import requests

if __name__ == '__main__':
    req_param = {
        "item": {
            "name": "Foo",
            "description": "The pretender",
            "price": 42.0,
            "tax": 3.2
        },
        "user": {
            "username": "dave",
            "full_name": "Dave Grohl"
        },
    }
    url = r"http://127.0.0.1:8000/items/12?importance=5"
    response = requests.request('PUT', url, json=req_param)
    print(response.json())
    # {'item_id': 12, 'item': {'name': 'Foo', 'description': 'The pretender', 'price': 42.0, 'tax': 3.2}, 'user': {'username': 'dave', 'full_name': 'Dave Grohl'}, 'importance': 5}

如果要将单类型查询参数 importance 从请求体传入，则可以使用 Body 指示 FastAPI 将其作为请求体的另一个键进行处理。

from typing import Annotated

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

app = FastAPI()


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


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


@app.put("/items/{item_id}")
async def update_item(
        item_id: Annotated[int, Path(title="The ID of the item to get", ge=10, le=1000)], item: Item, user: User, importance: Annotated[int, Body()]
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

import requests

if __name__ == '__main__':
    req_param = {
        "item": {
            "name": "Foo",
            "description": "The pretender",
            "price": 42.0,
            "tax": 3.2
        },
        "user": {
            "username": "dave",
            "full_name": "Dave Grohl"
        },
        "importance": 5
    }
    url = r"http://127.0.0.1:8000/items/12"
    response = requests.request('PUT', url, json=req_param)
    print(response.json())
    # {'item_id': 12, 'item': {'name': 'Foo', 'description': 'The pretender', 'price': 42.0, 'tax': 3.2}, 'user': {'username': 'dave', 'full_name': 'Dave Grohl'}, 'importance': 5}

嵌套模型

import uvicorn

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    req_param = {
        "name": "Looking",
        "description": "An optional description",
        "price": 2.3,
        "tax": 0.25,
        "tags": ["hello", "world"],
        "image": {
            "url": "http://example.com/baz.jpg",
            "name": "The Foo live"
        }
    }
    url = r"http://127.0.0.1:8000/items/12"
    response = requests.request('put', url, json=req_param)
    print(response.json())
    # {'item_id': 12, 'item': {'name': 'Looking', 'description': 'An optional description', 'price': 2.3, 'tax': 0.25, 'tags': ['hello', 'world'], 'image': {'url': 'http://example.com/baz.jpg', 'name': 'The Foo live'}}}

多层嵌套

import uvicorn

from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


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


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


class Offer(BaseModel):
    name: str
    description: str | None = None
    price: float
    items: list[Item]


@app.post("/offers/")
async def create_offer(offer: Offer):
    return offer

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    req_param = {
        "name": "Looking",
        "description": "An optional description",
        "price": 2.3,
        "items": [
            {
                "name": "item_name",
                "description": "item_description",
                "price": 3.4,
                "images": [
                    {
                        "url": "http://example.com/baz.jpg",
                        "name": "The Foo live"
                    },
                    {
                        "url": "http://example.com/dave.jpg",
                        "name": "The Baz"
                    }
                ]
            }
        ]
    }
    url = r"http://127.0.0.1:8000/offers"
    response = requests.request('post', url, json=req_param)
    print(response.json())
    # {
    #   'name': 'Looking',
    #   'description': 'An optional description',
    #   'price': 2.3,
    #   'items': [
    #     {
    #       'name': 'item_name',
    #       'description': 'item_description',
    #       'price': 3.4,
    #       'tax': None,
    #       'tags': [],
    #       'images': [
    #         {
    #           'url': 'http://example.com/baz.jpg',
    #           'name': 'The Foo live'
    #         },
    #         {
    #           'url': 'http://example.com/dave.jpg',
    #           'name': 'The Baz'
    #         }
    #       ]
    #     }
    #   ]
    # }

参数示例

可以使用 Config 和 schema_extra 为 Pydantic 模型声明一个示例

import uvicorn

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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,
                }
            ]
        }
    }


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


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

然后，就可以在文档中查看到了。

而且，在页面测试运行的时候，也可以直接在原有示例的基础上进行编辑测试。

数据更新

数据更新时，生成不含输入模型默认值的 dict （使用 exclude_unset 参数排除未显式指定的默认值）

import uvicorn

from typing import List, Union

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: Union[str, None] = None
    description: Union[str, None] = None
    price: Union[float, None] = 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.put("/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)  # 转换成 Item 模型
    update_data = item.model_dump(exclude_unset=True)  # 使用 exclude_unset 排除未显式指定的默认值
    updated_item = stored_item_model.model_copy(update=update_data)  # 创建副本并更新
    items[item_id] = jsonable_encoder(updated_item)  # 把模型副本转换为可存入数据库的形式并更新到 items
    return updated_item


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    req_param = {
        "name": "Looking",
        "description": "a test description",
        "price": 9.9
    }
    url = r"http://127.0.0.1:8000/items/bar"
    response = requests.request('put', url, json=req_param)
    print(response.json())
    # {'name': 'Looking', 'description': 'a test description', 'price': 9.9, 'tax': 20.2, 'tags': []}

Cookie 参数

参数声明

import uvicorn

from typing import Annotated

from fastapi import Cookie, FastAPI

app = FastAPI()


@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
    return {"ads_id": ads_id}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

cookie 没办法像路径参数和查询参数那样直接拼接在 url 中，所以只好写个请求来演示了。

import requests

if __name__ == '__main__':
    cookie = {
        "ads_id": "cookie_str"
    }
    url = r"http://127.0.0.1:8000/items"
    response = requests.request('get', url, cookies=cookie)
    print(response.json())  
    # {'ads_id': 'cookie_str'}

参数模型

import uvicorn

from typing import Annotated

from fastapi import Cookie, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Cookies(BaseModel):
    session_id: str  # 必选参数
    fatebook_tracker: str | None = None
    googall_tracker: str | None = None


@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
    return cookies

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

import requests

if __name__ == '__main__':
    cookie = {
        "session_id": "test_session_id",
        "fatebook_tracker": "fatebook_tracker",
        "googall_tracker": "googall_tracker"
    }
    url = r"http://127.0.0.1:8000/items"
    response = requests.request('get', url, cookies=cookie)
    print(response.json())
    # {'session_id': 'test_session_id', 'fatebook_tracker': 'fatebook_tracker', 'googall_tracker': 'googall_tracker'}

Header 参数

默认情况下，Header 把参数名中的字符由下划线 _ 改为连字符 - 来提取并存档请求头。HTTP的请求头不区分大小写。

参数声明

import uvicorn

from typing import Annotated

from fastapi import FastAPI, Header

app = FastAPI()


@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):
    return {"User-Agent": user_agent}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    header = {
        "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.6 (KHTML, like Gecko) Chrome/20.0.1092.0 Safari/536.6",
    }
    url = r"http://127.0.0.1:8000/items"
    response = requests.request('get', url, headers=header)
    print(response.json())
    # {'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.6 (KHTML, like Gecko) Chrome/20.0.1092.0 Safari/536.6'}

参数模型

import uvicorn

from typing import Annotated

from fastapi import FastAPI, Header
from pydantic import BaseModel

app = FastAPI()


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

    host: str
    save_data: bool
    if_modified_since: str | None = None
    traceparent: str | None = None
    x_tag: list[str] = []


@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
    return headers

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    header = {
        "Host": "127.0.0.1:8000",
        "Save-Data": "true",
        "If-Modified-Since": "if_modified_since",
        "Traceparent": "traceparent",
        "X-Tag": "hello"
    }
    url = r"http://127.0.0.1:8000/items"
    response = requests.request('get', url, headers=header)
    print(response.json())
    # {'host': '127.0.0.1:8000', 'save_data': True, 'if_modified_since': 'if_modified_since', 'traceparent': 'traceparent', 'x_tag': ['hello']}

响应模型

使用路径操作装饰器的 response_model 参数来定义响应模型，特别是确保私有数据被过滤掉。

输出限定

response_model 会将输出数据限制在该模型定义内。会负责过滤掉未在输出模型中声明的数据。

如下面的示例，添加限定时，返回结果自动将用户的密码信息去掉了。

import uvicorn

from typing import Any

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: str
    full_name: str | None = None


class UserOut(BaseModel):
    username: str
    email: str
    full_name: str | None = None


@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:
    return user

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

import requests

if __name__ == '__main__':
    req_param = {
        "username": "Looking",
        "password": "123456",
        "email": "looking@qq.com",
        "full_name": "lucky looking"
    }
    url = r"http://127.0.0.1:8000/user"
    response = requests.request('post', url, json=req_param)
    print(response.json())
    # {'username': 'Looking', 'email': 'looking@qq.com', 'full_name': 'lucky looking'}

如果不使用 response_model 进行限定，则会将用户的密码信息原样返回，十分危险。

import uvicorn

from typing import Any

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: str
    full_name: str | None = None


class UserOut(BaseModel):
    username: str
    email: str
    full_name: str | None = None


@app.post("/user/")
async def create_user(user: UserIn) -> Any:
    return user

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

import requests

if __name__ == '__main__':
    req_param = {
        "username": "Looking",
        "password": "123456",
        "email": "looking@qq.com",
        "full_name": "lucky looking"
    }
    url = r"http://127.0.0.1:8000/user"
    response = requests.request('post', url, json=req_param)
    print(response.json())
    # {'username': 'Looking', 'password': '123456', 'email': 'looking@qq.com', 'full_name': 'lucky looking'}

response_model

当模型定义了默认值时，响应模型会自动将缺失的值使用默认值进行补充（后面的 response_model_exclude_unset，response_model_include 和 response_model_exclude 都需要配合 response_model 使用，单独配置不生效）。

import uvicorn

from typing import List, Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    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]

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

`response_model_exclude_unset`

设置装饰器参数 response_model_exclude_unset=True ，然后响应中将不会包含那些默认值，而是仅有实际设置的值（显式指定的值，即使为空，也会展示）。

从 exclude unset 的拼写也可以看出，这个参数是为了排除掉没有显式指定的值。

import uvicorn

from typing import List, Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    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, response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items[item_id]

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

`response_model_include`

响应只包含指定的属性

import uvicorn

from typing import List, Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    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, response_model_include={"name", "description"})
async def read_item(item_id: str):
    return items[item_id]

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

response_model_exclude

响应排除指定的属性

import uvicorn

from typing import List, Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    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, response_model_exclude={"name", "description"})
async def read_item(item_id: str):
    return items[item_id]

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

状态码

100 及以上的状态码用于返回信息。这类状态码很少直接使用。具有这些状态码的响应不能包含响应体
200 及以上的状态码用于表示成功。这些状态码是最常用的
- 200 是默认状态代码，表示一切正常
- 201 表示已创建，通常在数据库中创建新记录后使用
- 204 是一种特殊的例子，表示无内容。该响应在没有为客户端返回内容时使用，因此，该响应不能包含响应体
300 及以上的状态码用于重定向。具有这些状态码的响应不一定包含响应体，但 304未修改是个例外，该响应不得包含响应体
400 及以上的状态码用于表示客户端错误。这些可能是第二常用的类型
- 404，用于未找到响应
- 对于来自客户端的一般错误，可以只使用 400
500 及以上的状态码用于表示服务器端错误。几乎永远不会直接使用这些状态码。应用代码或服务器出现问题时，会自动返回这些状态代码。

import uvicorn

from fastapi import FastAPI, status

app = FastAPI()


@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    url = r"http://127.0.0.1:8000/items?name=Looking"
    response = requests.request('post', url)
    print(response.status_code)  # 201

表单参数

如果接收的不是 JSON，而是表单字段时，要使用 Form。

参数声明

import uvicorn

from fastapi import FastAPI, Form

app = FastAPI()


@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

requests 请求时把请求数据放到 data 参数中。

import requests

if __name__ == '__main__':
    req_param = {
        "username": "Looking",
        "password": "123456"
    }
    url = r"http://127.0.0.1:8000/login"
    response = requests.request('post', url, data=req_param)
    print(response.json())
    # {'username': 'Looking'}

参数模型

为了屏蔽密码参数，设置 response_model_exclude 来排除 password 字段。

import uvicorn

from typing import Annotated

from fastapi import FastAPI, Form
from pydantic import BaseModel

app = FastAPI()


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


@app.post("/login/", response_model=FormData, response_model_exclude={"password"})
async def login(data: Annotated[FormData, Form()]):
    return data

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    req_param = {
        "username": "Looking",
        "password": "123456"
    }
    url = r"http://127.0.0.1:8000/login"
    response = requests.request('post', url, data=req_param)
    print(response.json())
    # {'username': 'Looking'}

文件参数

File

import uvicorn

from fastapi import FastAPI, File

app = FastAPI()


@app.post("/files/")
async def create_file(file: bytes = File()):
    return {"file_size": len(file)}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    file_data = [('file', open("test.txt", 'rb'))]
    url = r"http://127.0.0.1:8000/files"
    response = requests.request('post', url, files=file_data)
    print(response.json())  
    # {'file_size': 10947}

UploadFile

import uvicorn

from fastapi import FastAPI, UploadFile

app = FastAPI()


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    return {"filename": file.filename, "size": file.size}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    file_data = [('file', open("test.txt", 'rb'))]
    url = r"http://127.0.0.1:8000/uploadfile"
    response = requests.request('post', url, files=file_data)
    print(response.json())
    # {'filename': 'test.txt', 'size': 10947}

可选文件

可以通过使用标准类型注解并将 None 作为默认值的方式，将一个文件参数设为可选。

import uvicorn

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/files/")
async def create_file(file: bytes | None = File(default=None)):
    if not file:
        return {"message": "No file sent"}
    else:
        return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile | None = None):
    if not file:
        return {"message": "No upload file sent"}
    else:
        return {"filename": file.filename}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

import requests

if __name__ == '__main__':
    file_data = [('file', open("test.txt", 'rb'))]
    url = r"http://127.0.0.1:8000/uploadfile"
    response = requests.request('post', url, files=file_data)
    print(response.json())
    # {'filename': 'test.txt'}
    file_data = []
    url = r"http://127.0.0.1:8000/uploadfile"
    response = requests.request('post', url, files=file_data)
    print(response.json())
    # {'message': 'No upload file sent'}

元数据

元数据常应用于路由分组、接口描述、标签、说明、响应描述等方面，不会影响业务逻辑运行。

import uvicorn

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(description="A file read as UploadFile")):
    return {"filename": file.filename}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

多文件

import uvicorn

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile]):
    return {"filenames": [file.filename for file in files]}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    file_data = [('files', open("test.txt", 'rb')), ('files', open("test.xlsx", 'rb'))]
    url = r"http://127.0.0.1:8000/uploadfiles"
    response = requests.request('post', url, files=file_data)
    print(response.json())
    # {'filenames': ['test.txt', 'test.xlsx']}

元数据

在文件参数那块有提到过元数据。元数据常应用于路由分组、接口描述、标签、说明、响应描述等方面，不会影响业务逻辑运行。

API元数据

参数	类型	描述
`title`	`str`	API 的标题。
`summary`	`str`	API 的简短摘要。自 OpenAPI 3.1.0、FastAPI 0.99.0 起可用。.
`description`	`str`	API 的简短描述。可以使用Markdown。
`version`	`string`	API 的版本。这是您自己的应用程序的版本，例如 `2.5.0` 。
`terms_of_service`	`str`	API 服务条款的 URL。如果提供，则必须是 URL。
`contact`	`dict`	公开的 API 的联系信息。它可以包含多个字段。
`license_info`	`dict`	公开的 API 的许可证信息。它可以包含多个字段。

以上测试用的 FastAPI() 都没有添加额外的元数据，也可以也是可以按照如下添加元数据的。

import uvicorn

from fastapi import FastAPI

description = """
ChimichangApp API helps you do awesome stuff. 🚀

## Items

You can **read items**.

## Users

You will be able to:

* **Create users** (_not implemented_).
* **Read users** (_not implemented_).
"""

app = FastAPI(
    title="ChimichangApp",
    description=description,
    summary="Deadpool's favorite app. Nuff said.",
    version="0.0.1",
    terms_of_service="http://example.com/terms/",
    contact={
        "name": "Deadpoolio the Amazing",
        "url": "http://x-force.example.com/contact/",
        "email": "dp@x-force.example.com",
    },
    license_info={
        "name": "Apache 2.0",
        "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
    },
)


@app.get("/items/")
async def read_items():
    return [{"name": "Katana"}]


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

元数据页面展示效果如下，大家也可以自己比对一下元数据以及在页面的展示。

标签元数据 tags

将 tags 参数和路径操作一起使用，将其分配给不同的标签，还可以使用参数指定额外的文档链接。每个标签元数据字典的顺序也定义了在文档用户界面显示的顺序（如果没用标签 tags 显式指定，当然还是按照默认的代码中定义的先后顺序来）。

import uvicorn

from fastapi import FastAPI

tags_metadata = [
    {
        "name": "users",
        "description": "Operations with users. The **login** logic is also here.",
    },
    {
        "name": "items",
        "description": "Manage items. So _fancy_ they have their own docs.",
        "externalDocs": {
            "description": "Items external docs",
            "url": "https://fastapi.tiangolo.com/",
        },
    },
]

app = FastAPI(openapi_tags=tags_metadata)


@app.get("/items/", tags=["items"])
async def get_items():
    return [{"name": "wand"}, {"name": "flying broom"}]



@app.get("/users/", tags=["users"])
async def get_users():
    return [{"name": "Harry"}, {"name": "Ron"}]

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

展示效果

依赖项

依赖项就是一个函数，且可以使用与路径操作函数相同的参数。然后通过 Depends 将路径操作与这些依赖项函数建立联系，从而达到复用重复代码片段的效果。

函数依赖项

import uvicorn

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

而且，依赖项中的信息也可以在文档中展示

类依赖项

import uvicorn

from fastapi import Depends, FastAPI

app = FastAPI()

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=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


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

多级依赖项

import uvicorn

from typing import Union

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


def query_extractor(q: Union[str, None] = None):
    return q


def query_or_cookie_extractor(
    q: str = Depends(query_extractor),
    last_query: Union[str, None] = Cookie(default=None),
):
    if not q:
        return last_query
    return q


@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
    return {"q_or_cookie": query_or_default}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

路径依赖项

简单理解，就是将依赖项放到路径装饰器函数中。

import uvicorn

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: 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: 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"}]

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    header = {
        "X-Token": "fake-super-secret-token",
        "X-Key": "fake-super-secret-key"
    }
    url = r"http://127.0.0.1:8000/items"
    response = requests.request('get', url, headers=header)
    print(response.json())
    # [{'item': 'Foo'}, {'item': 'Bar'}]

全局依赖项

通过与定义与路径装饰器依赖项类似的方式，可以把依赖项添加至整个 FastAPI 应用（相同的依赖项放置的位置不同，所作用的范围也就不一样）。就可以为所有路径操作应用该依赖项（在对路径函数没做任何修改的情况下，就自动添加了响应的校验，所以这个功能简单理解就是一个强大的装饰器）。

import uvicorn

from fastapi import Depends, FastAPI, Header, HTTPException


async def verify_token(x_token: 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: str = Header()):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


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


@app.get("/items/")
async def read_items():
    return [{"item": "Portal Gun"}, {"item": "Plumbus"}]


@app.get("/users/")
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

import requests

if __name__ == '__main__':
    header = {
        "X-Token": "fake-super-secret-token",
        "X-Key": "fake-super-secret-key"
    }
    url = r"http://127.0.0.1:8000/items"
    response = requests.request('get', url, headers=header)
    print(response.json())
    # [{'item': 'Portal Gun'}, {'item': 'Plumbus'}]
    url = r"http://127.0.0.1:8000/users"
    response = requests.request('get', url, headers=header)
    print(response.json())
    # [{'username': 'Rick'}, {'username': 'Morty'}]

安全性

安全表单

在添加安全校验后，没有通过校验的接口无法正常请求。

import uvicorn

from typing import Annotated

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


@app.get("/items/")
async def read_items(token: Annotated[str, Depends(oauth2_scheme)]):
    return {"token": token}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

简单验证

import uvicorn

from typing import Union

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

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "alice@example.com",
        "hashed_password": "fakehashedsecret2",
        "disabled": True,
    },
}

app = FastAPI()


def fake_hash_password(password: str):
    return "fakehashed" + password


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None


class UserInDB(User):
    hashed_password: str


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def fake_decode_token(token):
    # This doesn't provide any security at all
    # Check the next version
    user = get_user(fake_users_db, token)
    return user


async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user


async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    user = UserInDB(**user_dict)
    hashed_password = fake_hash_password(form_data.password)
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

    return {"access_token": user.username, "token_type": "bearer"}


@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例

认证通过后，就可以通过校验正常请求其他接口了。

令牌验证

Passlib 封装了专门的密码哈希算法（如 Argon2、bcrypt、PBKDF2），并提供 自动管理策略、迁移机制，相比于标准库中的 hashlib，可以显著降低出错风险。

pip install pyjwt
pip install passlib
pip install argon2-cffi

原示例中的

from pwdlib import PasswordHash

用不了，换成了

from passlib.hash import argon2 as password_hash

以下是完整示例代码

import uvicorn

from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from passlib.hash import argon2 as password_hash
from pydantic import BaseModel

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$wagCPXjifgvUFBzq4hqe3w$CYaIb8sB+wtD+Vu/P4uod1+Qof8h+1g7bbDlBID48Rc",
        "disabled": False,
    }
}


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


class TokenData(BaseModel):
    username: str | None = None


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


class UserInDB(User):
    hashed_password: str


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return password_hash.verify(plain_password, hashed_password)


def get_password_hash(password):
    return password_hash.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except InvalidTokenError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token")
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")


@app.get("/users/me/", response_model=User)
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return [{"item_id": "Foo", "owner": current_user.username}]

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

使用令牌验证，只需要在用户首次时通过用户名密码进行校验，后续的接口请求，只要在令牌的有效期呢，都可以直接通过令牌进行请求。

当然，如果令牌已过期，就需要重新进行登录验证。

后台任务

import time

import uvicorn

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    time.sleep(10)
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例（请求发起以后，会立马返回响应结果，但是真正的日志写操作还在后台执行）

import requests

if __name__ == '__main__':
    url = r"http://127.0.0.1:8000/send-notification/test@qq.com"
    response = requests.request('post', url)
    print(response.json())
    # {'message': 'Notification sent in the background'}

使用 BackgroundTasks 也适用于依赖注入系统。可以在多个级别声明 BackgroundTasks 类型的参数。比如我们在路径函数中添加 BackgroundTasks，同时在查询参数中添加 BackgroundTasks。

import uvicorn
from typing import Annotated

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

请求示例1

import requests

if __name__ == '__main__':
    url = r"http://127.0.0.1:8000/send-notification/test@qq.com?q=test_query"
    response = requests.request('post', url)
    print(response.json())
    # {'message': 'Message sent'}

log.txt

notification for test@qq.com: some notification
found query: test_query
message to test@qq.com

请求示例2

import requests

if __name__ == '__main__':
    url = r"http://127.0.0.1:8000/send-notification/test@qq.com"
    response = requests.request('post', url)
    print(response.json())
    # {'message': 'Message sent'}

log.txt

notification for test@qq.com: some notification
found query: test_query
message to test@qq.com
message to test@qq.com

静态文件

可以使用 StaticFiles 从目录中自动提供静态文件。将本地目录挂载到指定路由，然后就可以通过路由链接直接访问本地目录的文件了。

import uvicorn

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/static", StaticFiles(directory="img"), name="static")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

多模块应用程序

正常来说，当应用程序很大的时候，会分成多个模块，各个模块的组员负责自己模块的开发，然后通过类似注册的方式，将模块在逻辑层面上再组合起来。从而提高代码的可维护性、复用性和团队协作效率。‌比如 Python 的 Flask 框架的 Blueprint，Go 的 Gin 框架的路由分离功能。

代码层级

.
├── app                  # 「app」是一个 Python 包
│   ├── __init__.py      # 这个文件使「app」成为一个 Python 包
│   ├── main.py          # 「main」模块，例如 import app.main
│   ├── dependencies.py  # 「dependencies」模块，例如 import app.dependencies
│   └── routers          # 「routers」是一个「Python 子包」
│   │   ├── __init__.py  # 使「routers」成为一个「Python 子包」
│   │   ├── items.py     # 「items」子模块，例如 import app.routers.items
│   │   └── users.py     # 「users」子模块，例如 import app.routers.users
│   └── internal         # 「internal」是一个「Python 子包」
│       ├── __init__.py  # 使「internal」成为一个「Python 子包」
│       └── admin.py     # 「admin」子模块，例如 import app.internal.admin

app/dependencies.py

依赖项一般各个模块都需要使用，所以放到外部

from fastapi import Header, HTTPException


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


async def get_query_token(token: str):
    if token != "jessica":
        raise HTTPException(status_code=400, detail="No Jessica token provided")

app/routers/users.py

可以看到，用户相关的路径中都统一加了 /users 前缀，用来与其他路由区分。

from fastapi import APIRouter

router = APIRouter()


@router.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]


@router.get("/users/me", tags=["users"])
async def read_user_me():
    return {"username": "fakecurrentuser"}


@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
    return {"username": username}

app/routers/items.py

但是我们也可以在 APIRouter 中通过设置 prefix 来统一设置相同前缀。

from fastapi import APIRouter, Depends, HTTPException

from app.dependencies import get_token_header

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)

fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}


@router.get("/")
async def read_items():
    return fake_items_db


@router.get("/{item_id}")
async def read_item(item_id: str):
    if item_id not in fake_items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"name": fake_items_db[item_id]["name"], "item_id": item_id}


@router.put(
    "/{item_id}",
    tags=["custom"],
    responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
    if item_id != "plumbus":
        raise HTTPException(
            status_code=403, detail="You can only update the item: plumbus"
        )
    return {"item_id": item_id, "name": "The great Plumbus"}

app/internal/admin.py

admin.py 表示我们无法直接修改的内部路由。

from fastapi import APIRouter

router = APIRouter()


@router.post("/")
async def update_admin():
    return {"message": "Admin getting schwifty"}

app/main.py

main.py 则是主函数，用于将各个模块的路由引入并组合起来

import uvicorn


from fastapi import Depends, FastAPI

from dependencies import get_query_token, get_token_header
from internal import admin
from routers import items, users

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


app.include_router(users.router)
app.include_router(items.router)
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)


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

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

api文档

请求示例

首先，我们一定要根据入参弄清楚 get_token_header 和 get_query_token 依赖项到底是针对什么类型参数的校验。这样，后续传参的时候才知道需要以什么样的方式传参。

from fastapi import Header, HTTPException


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


async def get_query_token(token: str):
    if token != "jessica":
        raise HTTPException(status_code=400, detail="No Jessica token provided")

token 的参数类型是 str 的简单类型，所以知道 token 是查询参数。
x_token 虽然也是 str 类型，但是指定了其为 Header()，所以是 Header 参数。

根路由

由于 get_query_token 被添加到整个应用的依赖中，所以每个路由请求都少不了对它的处理。根据 get_query_token 的定义，可知 token 是一个查询参数，所以仅仅需要拼接到路径中即可。

import requests

if __name__ == '__main__':
    url = r"http://127.0.0.1:8000/?token=jessica"
    response = requests.request('get', url)
    print(response.json())
    # {'message': 'Hello Bigger Applications!'}

admin.py

admin.py 的路由是通过下面这种方式注册的

app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)

因此，在请求时除了要传上面说的常规的 token，还要注意依赖中还有 X-Token 的校验。注意别忘了路由注册时添加的前缀 admin 哟。

import requests

if __name__ == '__main__':
    url = r"http://127.0.0.1:8000/admin?token=jessica"
    headers = {
        "X-Token": "fake-super-secret-token"
    }
    response = requests.request('post', url, headers=headers)
    print(response.json())
    # {'message': 'Admin getting schwifty'}

所以，到目前为止，我们知道了至少3种统一在路径种添加前缀的方式：

在路径操作手动添加统一前缀。比如示例中的 users.py
在初始化定义模块的 APIRouter 时设置 prefix 参数。比如示例种的 items.py
如果以上两种方式都错过了怎么办？还可以在 include_router 引入路由时添加 prefix。比如这里的 admin.py 中的路由注册方式

items.py

items.py 的路由 router 初始化时，用的下面的这种方式（也添加了 X-Token 校验，所以 token 和 X-Token 都需要传）。

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)

请求示例

import requests

if __name__ == '__main__':
    url = r"http://127.0.0.1:8000/items?token=jessica"
    headers = {
        "X-Token": "fake-super-secret-token"
    }
    response = requests.request('get', url, headers=headers)
    print(response.json())
    # {'plumbus': {'name': 'Plumbus'}, 'gun': {'name': 'Portal Gun'}}
    url = r"http://127.0.0.1:8000/items/plumbus?token=jessica"
    headers = {
        "X-Token": "fake-super-secret-token"
    }
    response = requests.request('get', url, headers=headers)
    print(response.json())
    # {'name': 'Plumbus', 'item_id': 'plumbus'}

users.py

users.py 中的路由实例化很简单，没有引用 get_token_header，所以请求时无需关注 X-Token 的 Header。

import requests

if __name__ == '__main__':
    url = r"http://127.0.0.1:8000/users?token=jessica"
    response = requests.request('get', url)
    print(response.json())
    # [{'username': 'Rick'}, {'username': 'Morty'}]
    url = r"http://127.0.0.1:8000/users/me?token=jessica"
    response = requests.request('get', url)
    print(response.json())
    # {'username': 'fakecurrentuser'}
    url = r"http://127.0.0.1:8000/users/guest?token=jessica"
    response = requests.request('get', url)
    print(response.json())
    # {'username': 'guest'}

测试

参数传递

不同类型的参数传递方式会有差异

传一个路径或查询参数，添加到URL上。
传一个JSON体，传一个Python对象(例如一个dict)到参数 json。
如果你需要发送 Form Data 而不是 JSON，使用 data 参数。
要发送 headers，传 dict 给 headers 参数。
对于 cookies，传 dict 给 cookies 参数。

postman

可以直接用接口调试工具 postman 发起请求进行测试。类似的工具还有 apiFox, apiPost 等。

requests

如果是简单的自测，可以像我示例中那样直接用 requests ，对于简单的 get 请求，甚至还可以直接用浏览器访问。

FastAPI

当然，也可以在 FastAPI 的在线文档中，使用 Try it out 测试。

TestClient

pip install httpx
pip install pytest

也可以使用 fastapi 中的 TestClient ，本质上是 starlette.testclient 的 TestClient，可以从 fastapi 下的 testclient.py 中窥见端倪：

from starlette.testclient import TestClient as TestClient  # noqa

结合 pytest，使用起来也很方便。

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()


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


client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}


if __name__ == '__main__':
    test_read_main()

数据库

FastAPI 不限定具体使用哪个数据库，一般像 Web 框架都有特有的 ORM 来进行封装（比如 Flask 的 SQLAlchemy），将不同数据库层面的语法差异屏蔽掉。

加入社区！打开量化的大门，首批课程上线啦！

更多推荐

终极指南：5步复现Spring Boot安全风险CVE-2016-1000027

Spring Boot安全风险CVE-2016-1000027是一个严重的Java反序列化问题，允许攻击者通过恶意序列化数据执行任意代码。本文将详细解析这一问题的原理、复现方法和修复方案。## 🔍 问题背景与原理剖析CVE-2016-1000027问题源于Spring Framework中的`HttpInvokerServiceExporter`和`RemoteInvocationSer

量化交易与投资社区

终极指南：如何避免RateLimitExceededException的Laravel API安全防护

在构建RESTful API时，避免RateLimitExceededException是每个开发者必须掌握的关键技能。Dingo API作为Laravel和Lumen框架的强大RESTful API包，提供了完整的速率限制机制来保护你的应用程序免受恶意请求和过量访问。本文将为你详细介绍如何配置和使用这些安全功能。## 🛡️ 什么是速率限制及其重要性速率限制是API安全的第一道防线，它能