基础使用

框架特色

优点

快速：可与 NodeJS 和 Go 比肩的极高性能（归功于 Starlette 和 Pydantic）
高效编码：提高功能开发速度约 200％至 300％
更少 bug：减少约 40％的人为（开发者）导致错误。
智能：极佳的编辑器支持。处处皆可自动补全，减少调试时间
简单：设计的易于使用和学习，阅读文档的时间更短
简短：使代码重复最小化。通过不同的参数声明实现丰富功能。bug 更少
健壮：生产可用级别的代码。还有自动生成的交互式文档
标准化：基于（并完全兼容）API 的相关开放标准：OpenAPI (以前被称为 Swagger) 和 JSON Schema。
支持ASGI

基础使用

官方文档

安装

# 主体
pip install fastapi

# 相关依赖安装
pip install uvicorn[standard] pydantic -i https://pypi.douban.com/simple/

运行

# 外部通过 uvicorn 运行
uvicorn app.py:function-name --reload --host=127.0.0.1 --port=4040

# 内部启动
if __name__ == "__main__":
      import uvicorn
    uvicorn.run(
      "app:app", 
      host="127.0.0.1", # 地址
      port=4040,        # 端口
      log_level="info", # 日志等级
      reload=True,      # 热更新
    )

依赖

from typing import Optional, Set
from fastapi import FastAPI, Query, Path, Cookie, Header
from pydantic import BaseModel

# 初始化服务
app = FastAPI()

# 定义一个 对象模型
# 可以用于请求的参数校验
# 可以用响应的的参数校验
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: List[str] = []

内置模块

from fastapi import <内置模块>

FastAPI - app实例化入口
Query - 验证请求入参
Path - 验证
Cookie - 验证 cookie
Header - 验证请求头部，请求参数等
Depends - 全局验证器
HTTPException - 错误处理，直接返回

from fastapi.responses import HTMLResponse

Depends

# 这里定义了一个验证 xtoken的函数
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")

# 这里定义了一个验证 x_key的函数
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"}]

内置入参

app.get/post(<内置入参>="xxxxx")

deprecated - 将路由设置为已废置

请求校验

Query

@app.get("/items/")
async def read_items(q: Union[str, None] = Query(
  default=None,         # 设置默认指，如果第一个参数不填，或者填 ... 则视为必须
  default=["foo", "bar"]# 变成可选，参数必须是列表中的值
  max_length=50,        # 限制最大长度
  min_length=3,         # 限制最小长度
  regex="^fixedquery$", # 使用正则校验
  title="Query string",
  description="xxxxxx",
  alias="xxx-xxx",      # 让参数支持非变量名的格式，如：cps-1, aaa-bbb
  deprecated=True,      # 标志这个参数激将废弃，文档会明确提示
  ge=1, # 大于等于
  le=1, # 小于等于
  gt=1, # 大于
  lt=1, # 小于
)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

Path

@app.get("/items/{item_id}")
async def read_items(
  item_id:str = Path(max_length = 32) # 对路径验证
  q: Union[str, None] = Query())      # 对参数验证

Body

Body 同样具有与 Query、Path 以及其他后面将看到的类完全相同的额外校验和元数据参数。

一个接口的入参指定了多个 BaseModel 时，将会被自动合并为body的key上
对于单个入参同时需要合并到body的，可以使用 xxxx = Body()来告诉FastAPI这是body的一部分

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

class User(BaseModel):
    username: str
    full_name: Union[str, None] = None
      
@app.put("/items/{item_id}")
async def update_item(
  item_id: int, 
  item: Item,   # 当前指定了多个 BaseModel
  user: User, 
  importance: int = Body() # 告诉fastapi 我也是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
}

`xxx = Body(embed=True)` 展开请求体到 body

// 在这种情况下，**FastAPI** 将期望像这样的请求体：
{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}
// 而不是：
{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2
}

请求相关

基本使用

# main.py
from typing import Optional
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
    return {"item_id": item_id, "q": q}

uvicorn main:app --reload # 启动服务
http://127.0.0.1:8000/items/5?q=somequery
http://127.0.0.1:8000/docs  # 自动生成swagger
http://127.0.0.1:8000/redoc # 自动生成ReDoc

响应相关

响应数据 `response_model`

使用response_model参数来声明用于响应的模型

response_model_exclude_unset=True：响应中将不会包含那些默认值，而是仅有实际设置的值
response_model_include包含哪些属性
response_model_exclude省略某些属性

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

状态码 `status_code`

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

基础示例

参数校验

Query

@app.get("/items/")
# 使用 fastapi.Query 对q 进行校验
# min_length 最小长度
# regex 正则匹配
# Query 第一个参数为默认值，...表示是必需的
async def read_items(q: Optional[str] = Query(None, max_length=50)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

Path

@app.get("/items/{item_id}")
# 使用 fastapi.Path 对参数进行校验
# gt：大于
# ge：大于等于
# lt：小于
# le：小于等于
async def read_items(
    *,
    item_id: int = Path(..., title="The ID of the item to get", ge=0, le=1000),
    q: str,
    size: float = Query(..., gt=0, lt=10.5)
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

Cookie

@app.get("/items/")
# 使用 fastapi.Cookie 对参数是否一个cookie作基本校验
async def read_items(ads_id: Optional[str] = Cookie(None)):
    return {"ads_id": ads_id}

Header

@app.get("/items/")
# 使用 fastapi.Header 对参数是否一个Header作基本校验
async def read_items(user_agent: Optional[str] = Header(None)):
    return {"User-Agent": user_agent}

路由分组

通过在路由的入参tags-tagsName，将路由进行分组，该分组会体现在fastapi生成的swagger或者redoc上面

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

接口文档

ummary 、 description

通过fastapi.summary 和fastapi.description明确该接口的文档

@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

多行注释

@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

废弃路由

@app.get("/elements/", tags=["items"], deprecated=True)
async def read_elements():
    return [{"item_id": "Foo"}]

跨域设置

具体参数

llow_origins - 一个允许跨域请求的源列表。例如 ['https://example.org', 'https://www.example.org']。你可以使用 ['*'] 允许任何源。

allow_origin_regex - 一个正则表达式字符串，匹配的源允许跨域请求。例如 'https://.*\.example\.org'。

allow_methods - 一个允许跨域请求的 HTTP 方法列表。默认为 ['GET']。你可以使用 ['*'] 来允许所有标准方法。

allow_headers - 一个允许跨域请求的 HTTP 请求头列表。默认为 []。你可以使用 ['*'] 允许所有的请求头。Accept、Accept-Language、Content-Language 以及 Content-Type 请求头总是允许 CORS 请求。

allow_credentials - 指示跨域请求支持 cookies。默认是 False。另外，允许凭证时 allow_origins 不能设定为 ['*']，必须指定源。

expose_headers - 指示可以被浏览器访问的响应头。默认为 []。

max_age - 设定浏览器缓存 CORS 响应的最长时间，单位是秒。默认为 600。

你可以在FastAPI应用中使用CORSMiddleware来配置跨域：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()

# 设置支持需要支持的跨域网段
origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

# 通过 add_middleware 开启跨域
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

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

路由配置

枚举路由

通过枚举Enum定义的路由，在生成对应的swagger时也会被自动的填入可选范围

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

app = FastAPI()

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    if model_name == 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"}

数据模型

请求头数据

通过定义 Config 和 schema_extra，可以定义对应模型的示例对象，同时会反应到文档中

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

    # 定义一个示例
    class Config:
        schema_extra = {
            "example": {
                "name": "Foo",
                "description": "A very nice Item",
                "price": 35.4,
                "tax": 3.2,
            }
        }

参阅文献

https://juejin.cn/post/7075320308483391519#heading-2

框架特色​

优点​

基础使用​

官方文档​

安装​

运行​

依赖​

内置模块​

Depends​

内置入参​

请求校验​

相关依赖​

Query​

Path​

Body​

xxx = Body(embed=True) 展开请求体到 body​

请求相关​

基本使用​

响应相关​

响应数据 response_model​

状态码 status_code​

基础示例​

参数校验​

路由分组​

接口文档​

ummary 、 description​

多行注释​

废弃路由​

跨域设置​

具体参数​

路由配置​

枚举路由​

数据模型​

请求头数据​

参阅文献​

框架特色

优点

基础使用

官方文档

安装

运行

依赖

内置模块

Depends

内置入参

请求校验

相关依赖

Query

Path

Body

`xxx = Body(embed=True)` 展开请求体到 body

请求相关

基本使用

响应相关

响应数据 `response_model`

状态码 `status_code`

基础示例

参数校验

路由分组

接口文档

ummary 、 description

多行注释

废弃路由

跨域设置

具体参数

路由配置

枚举路由

数据模型

请求头数据

参阅文献