处理请求

查询参数类Query

概述

FastAPI提供了查询参数类Query，几种使用方式如下：

添加单个约束条件
同时使用多个约束条件
在约束中使用正则表达式
在Query类中使用必选查询参数
使用列表泛型的查询参数
参数对象元数据
参数别名
弃用参数

添加单个约束条件

例如，我们可以对查询参数q添加约束条件，可选、类型str、参数值的长度不能超过10个字符。

在下文的例子中，使用q: Optional[str] = Query(None, max_length=10)替换了默认的查询参数定义方式，其作用是：

定义可选查询参数q，类型为str。
通过q的默认值调用Query类，生成Query类的实例，其主要作用是将q明确定义为查询参数，和《5.FastAPI [1/3]》里使用Body类将参数定义为请求体的方式是一样的。
Query类的第一个参数是None，设置查询参数q的默认值为None。
Query类的第二个参数是max_length=10，设置查询参数q的最大长度是10。

示例代码：

from typing import Optional
# 导入Query类
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()


@app.get("/")
# 使用Query定义规则
async def read_items(q: Optional[str] = Query(None, max_length=10)):
    return {"q": q}


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

同时使用多个约束条件

使用Query类作为参数默认值时，允许添加多条规则。

例如，为查询参数q增加一条最小长度的规则。
在下文的例子中，在Query类的参数中增加了一条min_length=3，作用是限制参数q的最小长度是3个字符。

示例代码：

from typing import Optional
# 导入Query类
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()


@app.get("/")
# 使用Query定义规则
async def read_items(q: Optional[str] = Query(None, min_length=3, max_length=10)):
    return {"q": q}


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

在约束中使用正则表达式

在下文的例子中，在Query类的参数中增加了一个参数regex，是一个正则表达式。

示例代码：

from typing import Optional
# 导入Query类
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()


@app.get("/")
# 使用Query定义规则
async def read_items(q: Optional[str] = Query(None, regex='^[\w\d]{3,10}$')):
    return {"q": q}


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

在Query类中使用必选查询参数

通过《5.FastAPI [1/3]》的讨论，我们知道，在查询参数定义时没有参数值，则为必选查询参数，例如用q:str代替q:str =None。

但是当使用Query类约束查询参数时，为了保证查询参数为必选，则需要把None替换为...。例如：

1	async def read_items(g: Optional [str] = Query(..., min_length=3)):

使用列表泛型的查询参数

格式如下:

1	查询参数:Optional[List[数据类型]]=Query(None)

在下文的例子中，查询参数q定义为可选参数，其类型是列表泛型List，值类型为str。

然后我们可以通过类似的方式请求。

1	http://127.0.0.1:8000/?q=a&q=b&q=c

示例代码：

from typing import List, Optional
import uvicorn
from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/")
# 定义列表类型的查询参数
async def read_items(q: Optional[List[str]] = Query(None)):
    return {"q": q}


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

我们还可以为参数列表指定默认值，修改代码中的Query类的列表默认值，示例代码：

1	async def read_items(q: List[str] = Query(["a", "b", "c"]))

另外，使用列表泛型接收查询参数时，也可以直接使用list代替List[str]，在这种情况下，FastAPI将不会检查列表泛型的值类型。

参数别名

假设存在一个URL如下：

1	http://127.0.0.1:8000/items/?item-query=abc

参数名为item-query，不是一个有效的Python变量名，FastAPI无法将这个参数名转换有效的查询参数。
这时，可以在Query中设置alias参数，给item-query设置一个别名。

示例代码：

1	async def read_items(q: Optional[str] = Query(None, alias="item-query")):

当设置了Query的参数alias="item-query"后，FastAPI在解析URL中的参数item-query=abc时，会将这个参数的值传给查询参数q，这样就避免了无效变量名的问题。

参数对象元数据

描述信息

Query类不但可以为查询参数做一些额外的校验，还可以添加更多其他信息，这些信息也会包含在生成的API文档中，供API文档页面显示和其他外部工具所使用，这些信息称为"元数据"。

例如，给参数添加标题和描述信息。

在下文的代码中，查询参数的默认值为Query类实例，其参数中添加了字符串校验规则，并且增加了元数据description，用于显示描述信息。

示例代码：

from typing import Optional
import uvicorn
from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
# 定义查询参数q，默认值为Query类
# 可选参数默认值None
# 定义规则
# 参数详细说明
async def read_items(q: Optional[str] = Query(None, min_length=3, description="根据此参数查找匹配的数据", )):
    return {"q": q}


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

弃用参数

在有些情况下，随着我们接口的迭代，有一些参数可能会被废弃，但为了兼容性，还需要再保留一段时间。这时候，希望在文档上将其展示为"已弃用"，可以在Query的参数中设置deprecated=True。

示例代码：

from typing import Optional
import uvicorn
from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
# 定义查询参数q，默认值为Query类；可选参数默认值None；定义规则；参数详细说明
# 定义查询参数name；可选参数默认值None；标记为弃用；参数说明
async def read_items(q: Optional[str] = Query(None, min_length=3, description="根据此参数查找匹配的数据", ),
                     name: Optional[str] = Query(None, deprecated=True, description="按名称查询")):
    return {"q": q, "name": name}


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

路径参数类Path

概述

路径参数类Path和查询参数类Query的用法很相似，Path类可以为路径参数设置校验规则和添加元数据，两者都继承了Param类。

例子

添加元数据

示例代码：

import uvicorn
# 导入Path类
from fastapi import FastAPI, Path

app = FastAPI()


@app.get("/items/{item_id}")
# 定义路径参数，设置默认值为Path函数
# 路径参数是必选参数
# 设置描述信息
async def read_items(item_id: int = Path(..., description="项目ID是路径的一部分"), ):
    return {"item_id": item_id}


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

数值校验规则：大于、小于、等于

示例代码：

from fastapi import FastAPI, Path
import uvicorn

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(item_id: int = Path(..., description="某一项的ID", ge=1),):
    return {"item_id": item_id}


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

数值校验可使用的参数如下：

gt，大于(greater than)
ge，大于等于(greater than or equal)
lt，小于(less than)
le，小于等于(less than or equal)。

Cookie参数类

在FastAPI中，可以通过Cookie参数类解析请求中的Cookie的值。

示例代码

from typing import Optional
import uvicorn
from fastapi import Cookie, FastAPI

app = FastAPI()


@app.get("/items/")
# 定义Cookie参数，默认为空
async def read_items(user_id: Optional[str] = Cookie(None)):
    return {"user_id": user_id}


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

在FastAPI中，可以通过Header参数类处理HTTP的Header。

在下文的例子中，在路径函数read_items中定义了一个可选Header参数user_agent，类型为str，默认值为None。

FastAPI在接收Header数据时，会自动把连字符-转换为下划线_，所以会通过user_agent去User-Agent的值。

示例代码：

from typing import Optional
import uvicorn
from fastapi import FastAPI, Header

app = FastAPI()


@app.get("/items/")
# 定义Header参数，类型str，默认值为空
async def read_items(user_agent: Optional[str] = Header(None)):
    return {"User-Agent": user_agent}


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

特别的，如果我们不需要将连字符-转换为下划线_，可以设置Header函数的参数convert_underscores=False。

1	async def read_items(user_agent: Optional[str] = Header(None, convert_underscores=False)):

Pydantic的Field类

数据模型字段的规则设置示例

在下文的例子中，从Pydantic包导入了Field类，然后在定义数据模型时，description字段和price字段的默认值都调用了Field类，进行字段规则设置。

示例代码：

from typing import Optional
import uvicorn
from fastapi import Body, FastAPI
# 导入Field函数
from pydantic import BaseModel, Field

app = FastAPI()


# 定义数据模型类、继承自BaseModel
class Item(BaseModel):
    name: str
    description: Optional[str] = Field(
        None,
        title="一大段说明信息",
        max_length=300, )

    price: float = Field(
        ...,
        gt=0,
        description="单价必须大于0")
    tax: Optional[float] = None


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


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

配置类设置统一的元数据

业可以通过配置类Config统一设置数据模型的元数据。

在下文的例子中，在定义数据模型时，增加了一个配置类Config，Config给数据模型定义了一些样例数据，这些样例数据会在API文档中显示。

示例代码：

from typing import Optional
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


# 定义数据模型类
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

    # 定义配置类，必须命名为Config
    class Config:
        # 元数据
        # 定义元数据中的示例数据
        schema_extra = {
            "example": {
                "name": "三酷猫",
                "description": "这是一个非常不错的项目",
                "price": 35.4,
                "tax": 3.2,
            }
        }


# 定义路径参数
@app.put("/items/{item_id}")
# 路径参数和Body参数
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "item": item}


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

当然，我们也可以使用Field类设置样例数据，示例代码：

# 定义数据模型
class Item(BaseModel):
    # 使用 Field 对象的 example参数
    name:str = Field(..., example="kaka")
    description: Optional[str] = Field(None, example="这是一个不错的项目”)
    price: float = Field(..., example=35.4)
    tax: Optional[float] = Field(None, example=3.2)

另外，我们在使用Body类中也可以使用example参数定义样例数据。

复杂的请求数据模型

例如，在数据模型Item增加一个tags字段，记录标签列表。

示例代码：

from typing import Optional, List
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


# 定义数据模型类
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: List[str] = []


# 定义路径参数
@app.put("/items/{item_id}")
# 路径参数和Body参数
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "item": item}


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

此外，还可以在数据模型中嵌套数据模型，数据模型与泛型的结合使用等，与《5.FastAPI [1/3]》的讨论没有区别，这里不赘述。

任意类型的请求体

在有些情况下，有时候请求数据是列表类型，或者虽然是键值对结构，但请求数据中的键不是固定的。

第一种情况，对于列表类型的请求数据，可以直接用List来接收数据。示例代码：

from typing import List
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


# 定义数据模型
class Image(BaseModel):
    url: str
    name: str


@app.post("/images/")
# 定义请求体，列表泛型
async def create_multiple_images(images: List[Image]):
    return images


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

第二种情况，请求数据的结构是非固定格式的键值对，无法通过Pydantic库定义为数据模型，此时可以使用Python的数据类型Dict来接收数据。示例代码：

from typing import Dict
import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.post("/scores/")
# 使用字典泛型接收键值对数据
async def create_scores(scores: Dict[int, float]):
    return scores


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

直接使用请求类

在某些情况下，需要直接使用请求类(Request)，不需要对数据进行校验和转换。

例如，在路径操作函数中直接获取客户机的IP地址时。
在下文的例子中，在路径操作函数的参数中定义了请求类(Request)的实例request，然后可以通过request使用请求实例的各种属性。

示例代码：

from fastapi import FastAPI, Request
import uvicorn

app = FastAPI()


# 定义路径参数
@app.get("/host/")
# 直接使用Request对象
def read_root(request: Request):
    client_host = request.client.host
    return {"客户端主机地址": client_host}


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

定义响应

自定义Cookie数据

在响应中自定义Cookie数据的方式为，在路径操作函数中创建响应类的实例，使用该实例的set_cookie方法设置Cookie内容，最后返回这个响应类的实例即可。

在下文的例子中，定义了一个路径操作函数create_cookie，在其中创建响应类JSONResponse的实例response，然后通过response.set_cookie()方法，设置Cookie内容，最后返回带Cookie内容的响应数据。

示例代码：

from fastapi import FastAPI
from fastapi.responses import JSONResponse
import uvicorn

app = FastAPI()


@app.post("/cookie/")
def create_cookie():
    # 创建响应数据
    content = {"message": "threecoolcats like cookies"}
    # 创建响应类实例
    response = JSONResponse(content=content)
    # 设置Cookie
    response.set_cookie(key="user_id", value="9527")
    return response


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

自定义Header数据的方式和自定义Cookie数据类似，在路径操作函数中创建响应类的实例，在创建响应类实例时，以字典的形式传入要返回的Header参数。

在下文的例子中，在创建响应类JSONResponse实例时，传入了两个参数，第一个是响应的content内容数据，第二个是响应的自定义字典型Header数据。

示例代码：

from fastapi import FastAPI
from fastapi.responses import JSONResponse
import uvicorn

app = FastAPI()


# 注册路由路径
@app.get("/headers/")
# 创建路径操作函数
def get_headers():
    content = {"message": "Hello"}
    headers = {"X-a-b-c": "a-b-c",
               "User-Agent": "XXX Browser"}
    response = JSONResponse(content=content, headers=headers)
    return response


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

另外，在响应中设置Header数据时，有以下三种约定：

设置内置的Header(比如User-Agent、Content-Type)时，可以直接设置该Header对应的数据。
设置非设置的Header，名称需要以X开头。
例如，在要响应的Header中返回响应时间，可以将Header名称设置为"X-ResponseTime"，但不要设置为"Response-Time"。
增加自定义Header时，自定义Header的名称和内容都不能包含下划线_。

自定义响应状态码

在某些业务场景下，需要修改服务端响应的默认状态码，可以通过路径操作函数的装饰器实现修改默认状态码的功能。

在下文的例子中，在路径操作函数的装饰器的参数中，设置了status_code=201，其作用是将本次响应的状态码设置为201。如果没有设置这个参数，本次响应返回成功的状态码是200。

示例代码：

from fastapi import FastAPI, status
import uvicorn

app = FastAPI()


# 注册路由路径，设置状态码
@app.get("/items/", status_code=201)
# 定义路径操作函数
async def create_item(name: str):
    return {"name": name}


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

在FastAPI里提供了status模块，该模块中定义了所有响应状态码的别名。比如以上示例中，将status_code参数设置为

1	status_code = status.HTTP_201_CREATED

异常处理

异常类HttpException

在FastAPI中，使用HttpException异常类来处理异常信息，通过raise关键字来主动抛出异常信息。

在下文的例子中，导入HttpException异常类，在路径操作函数中判断item_id的值是否在模拟数据中，如果不在模拟数据中，则抛出一个HttpException类型的异常，并且在HTTPException中定义headers内容。

示例代码：

from fastapi import FastAPI, HTTPException
import uvicorn

app = FastAPI()

# 定义模拟数据
items = {"1": "cat"}


# 注册路由路径，定义路径参数
@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        # 使用raise抛出异常
        raise HTTPException(status_code=404, detail="未找到指定项目", headers={"X-Error": "访问项目出错"})
    return {"item": items[item_id]}


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

全局异常处理器

FastAPI提供了一种全局异常处理器的方式，通过自定义不同类型的异常，将逻辑处理代码与异常处理代码完全分开。

在下文的例子中，实现了一个全局异常处理器，其关键步骤如下：

自定义异常MyException类，继承自Python中的内建异常类Exception，并且重写了构造方法。
定义异常处理函数，函数的参数为请求类实例和异常类实例，在函数中返回了响应类JSONResponse的实例，其中的参数为响应状态码和异常信息，使用装饰器@app.exception_handler将异常处理函数注册为全局异常处理器。
定义路径操作函数，在函数中主动抛出自定义异常。

示例代码：

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

app = FastAPI()


# 定义异常类，继承自Exception
class MyException(Exception):
    def __init__(self, name: str):
        self.name = name


# 注册全局异常管理器
@app.exception_handler(MyException)
# 定义异常处理函数
async def my_exception_handler(
        # 请求类实例
        request: Request,
        # 异常类实例
        exc: MyException):
    # 返回响应类实例
    return JSONResponse(
        # 响应状态码
        status_code=418,
        # 状态文本
        content={"message": f"OMG，{exc.name}又迷路了"},
    )


# 注册路由路径，定义路径参数
@app.get("/cats/{name}")
async def find_cats(name: str):
    if name == "三酷猫":
        # 抛出自定义异常
        raise MyException(name=name)
    return {"cat": name}


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

内置异常处理器

FastAPI内置了一些异常类，比如验证请求数据时，数据无效则会引发异常出错RequestValidationError。

FastAPI为这些异常类提供了内置的异常处理器，当然有时我们也需要改变这些内置的异常的消息内容或格式。

在下文的例子中，重新注册了2个异常处理器，并且分别重新写了异常处理器函数的代码逻辑。在路径操作函数中，会引发两个异常，这两个异常会被重新注册的异常处理器分别捕获并处理。

示例代码：

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
# 导入普通文本响应类
from fastapi.responses import PlainTextResponse
# 使用别名导入
from starlette.exceptions import HTTPException as StarletteHTTPException
import uvicorn

app = FastAPI()


# 注册系统异常StarletteHTTPException
@app.exception_handler(StarletteHTTPException)
# 覆盖系统异常处理器，重写方法实现
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


# 注册系统异常RequestValidationError
@app.exception_handler(RequestValidationError)
# 覆盖系统异常处理器，重写方法
async def validation_exception_handler(request, exc):
    return PlainTextResponse(str(exc), status_code=400)


# 注册路由路径，定义路径参数
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="禁止使用3")
    return {"item_id": item_id}


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

解释说明：该例使用了Starlette库中的HTTPException类，Starlette是FastAPI使用的底层框架库，在FastAPI中可以直接使用Starlette库中的基础类，FastAPI的异常类HttpException是封装的Starlette中的异常类HTTPException。

在有些情况下，在处理完自定义异常后，还需要将异常处理器还原到默认状态，这时可以导入FastAPI内置的异常处理器，并在异常处理函数中重新调用内置的异常处理器，使FastAPI的异常处理功能还原到默认状态。

主要两点：

导人系统内置的异常处理器
在自定义异常处理函数中，完成其他操作后，使用return await调用内置异常处理器。

示例代码：

# 导人内置异常处理器
from fastapi.exception_handlers import(
    http_exception_handler, 
    request_validation_exception_handler, )

# 注册异常处理器
@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):
    print(f"OMG!网络错误!:{repr(exc)}")
    # 自定义代码处理完成后,还原为默认的异常处理器
    return await http_exception_handler(request, exc)

@app.exception_handler(RequestValidationError)
async def validation_ exception handler(request, exc):
    print(f"OMG! 请求数据无效!:{exc}")
    # 自定义代码处理完成后,还原为默认的异常处理器
    return await request_validation_exception_handler(request, exc)

中间件技术

中间件，Middleware，在每个请求处理之前被调用，又在每个响应返回给客户端之前被调用。

类似于Java中的过滤器和拦截器。

关于过滤器，可以参考《基于Java的后端开发入门：13.Servlet、Filter和Listener》的"Filter"部分。
关于拦截器，可以参考《基于Java的后端开发入门：17.SpringMVC》的"拦截器"部分。

在大多数情况下，FastAPI自带的中间件能够自动实现数据的接收和发送功能，无需额外处理。

但我们业可以调用中间件或自定义中间件功能，去实现某些特殊的功能。

自定义中间件

自定义中间件的过程，先定义一个中间件函数，然后在这个函数上增加装饰器@app.middleware("http")，该函数的参数包括了请求类Request的实例request和处理过程回调函数参数call_next。

在下文的例子中，实现了一个完整的自定义中间件，中间件函数的内部逻辑是在接收到请求以后，记录一个时间点，等待响应结束后，再计算出处理响应的时间差，并将这个时间差记录到Header中，返回给客户端。这个操作是全局的，Web服务器端对接收到的每个请求都会调用中间件执行操作。

示例代码：

import time
import uvicorn
from fastapi import FastAPI, Request

app = FastAPI()


# 使用装饰器，将函数注册为中间件函数
@app.middleware("http")
# 定义中间件函数，包含两个参数
async def add_process_time_header(request: Request, call_next):
    # 此处在路径操作收到请求之前
    # 记录时间点1
    start_time = time.time()
    # 获取响应类实例
    response = await call_next(request)
    # 此处在生成响应数据但返回之前
    # 计算处理时间
    process_time = time.time() - start_time
    # 修改响应Header
    response.headers["X-Process-Time"] = str(process_time)
    # 返回响应实例
    return response


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

CORSMiddleware

CORS， Cross-Origin Resource Sharing，跨域资源共享，简称跨域。

关于跨域的更多，可以参考我们在《关于弹幕视频网站的例子：基于Serverless的弹幕视频网站实现方案》中的讨论，这里不赘述。

FastAPI提供了CORS中间件，CORSMiddleware，用于处理跨域资源共享的问题。

在下文的例子中，导入了中间件CORSMiddleware，定义了一个可用域列表，接着用app.add_middleware()方法将中间件添加到FastAPI应用中。该方法的第一个参数指定需要加入的中间件类，其余参数是CORSMiddleware中间件可用的配置项。

示例代码：

from fastapi import FastAPI
# 导入CORSMiddleware
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

app = FastAPI()

# 定义可用域列表
origins = [
    "http://kakawanyifan.com",
    "https://kakawanyifan.com",
    "http://localhost",
    "http://localhost:8080",
]

# 在应用上添加中间件
app.add_middleware(
    # 内置中间件类
    CORSMiddleware,
    # 可用域列表
    allow_origins=origins,
    # 允许cookie， 是
    allow_credentials=True,
    # 参数3 允许的方法， 全部
    allow_methods=["*"],
    # 参数4 允许的Header，全部
    allow_headers=["*"],
)


# 注册路由路径
@app.get("/")
async def main():
    return {"message": "Hello World"}


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

CORSMiddleware支持以下参数：

allow_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。

HTTPSRedirectMiddleware

FastAPI提供了HTTPSRedirectMiddleware中间件，强制使用HTTS协议访问服务器端，对于任何传入的以HTTP开头的请求，都将被重新定向到HTTPS开头的请求

在下文的例子中，先导入了HTTPSRedirectMiddleware中间件，然后使用app.add_middleware()方法，将该中间件添加到应用上。

示例代码：

from fastapi import FastAPI
# 导入中间件
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
import uvicorn

app = FastAPI()

# 添加中间件，无其他参数
app.add_middleware(HTTPSRedirectMiddleware)


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


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

TrustedHostMiddleware

FastAPI提供了TrustedHostMiddleware中间件，配合app.add_middleware()的allowed_hosts参数，可以设置域名访问白名单。

在下文的代码中，先导入了中间件TrustedHostMiddleware，然后使用app.add_middleware()方法添加到应用上，同时设置了一个参数allowed_hosts列表，该列表设置可访问主机域名地址，如果想要任意主机域名地址都可以访问, 可以将参数设置为allowed_hosts=["*"]，或者不使用这个中间件。如果调用者传入请求的主机域名地址不在名单内，中间件将会给调用者返回一个400响应，表示资源不可用。

示例代码：

from fastapi import FastAPI
# 导入中间件
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import uvicorn

app = FastAPI()

# 添加中间件
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"])


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


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

GZipMiddleware

GZipMiddleware，用于压缩响应数据，其作用是当客户端向服务器端发的请求Header中带有"Accept-Encodin:GZip"时，对响应的数据以GZip格式进行压缩后，再发送给客户端(浏览器)。客户端(浏览器)接收到压缩数据后，先将数据解压缩，再解析。

该中间件有一个参数minimum_size，其作用是设置数据包的最小值，也就是说，只有当需要传递的数据长度大于这个值时，才会使用GZip压缩，否则将会传递原始数据。

示例代码：

from fastapi import FastAPI
# 导入中间件
from fastapi.middleware.gzip import GZipMiddleware
import uvicorn

app = FastAPI()

# 添加中间件，设置压缩参数
app.add_middleware(GZipMiddleware, minimum_size=1000)


@app.get("/")
async def main():
    return "somebigcontent"


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

依赖注入

概述

关于什么是依赖注入，可以参考《基于Java的后端开发入门：15.Spring Framework [1/2]》。

本文主要讨论在FastAPI中的实现方法。

需要注意的是，和Java中的依赖注入不一样。在Java中，我们可以在任意位置调用被注入的对象，但是在FastAPI中，只能在开头调用。在实际体验中，可能和AOP更像，尤其是"前置通知"。

使用函数实现依赖注入

在FastAPI中，先定义依赖项函数，然后在路径操作函数上定义它需要使用的依赖项。

在运行时，FastAPI根据路径操作函数的需求提供依赖项，称为"注入"依赖项。在以下场景中，使用依赖注入的方式可以减少代码重复和逻辑耦合：

共享逻辑(反复使用相同的代码逻辑)
共享数据库连接
加强安全，身份验证、角色需求等。

FastAPI通过自带的Depends()方法指定依赖函数来实现函数的依赖注入。

在下文的例子中，定义了一个依赖函数dep_params，此函数接收路径操作函数的所有参数，对参数进行处理，然后返回；在路径操作函数中，通过Depends()方法指定了依赖函数dep_params，而不是直接调用dep_params，这就是在FastAPI中使用依赖注入的方式。

示例代码：

from typing import Optional
import uvicorn
from fastapi import Depends, FastAPI

app = FastAPI()


# 定义依赖函数
async def dep_params(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


# 注册路由路径
@app.get("/items/")
# 定义路径操作函数
# 参数中定义依赖项
async def read_items(commons: dict = Depends(dep_params)):
    return commons


# 注册路由路径
@app.get("/users/")
# 定义路径操作函数
# 参数中定义依赖项
async def read_users(commons: dict = Depends(dep_params)):
    return commons


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

依赖函数的写法和路径操作函数一样，我们可以把它看作一个没有装饰器的路径操作函数，也可以返回任何类型的数据。

当Web服务器端接收到一个请求后，FastAPI的内部操作如下：

对URL进行路由匹配，调用对应的路径操作函数。
在路径操作函数里调用由Depends()指定的依赖函数dep_params。
从依赖函数中获取结果。
将该结果返回给路径操作函数中的定义的返回数据对象。

使用类实现依赖注入

除了通过依赖函数的方式实现依赖注人，也可以通过依赖类完成相同的依赖注入功能。

在下文的例子中，定义了依赖类，没有使用依赖函数，将多个参数封装成一个参数类，不但增加了代码的可读性，还可以在开发工具上使用代码提示和开发环境的自动完成等辅助功能。

示例代码：

from typing import Optional
import uvicorn
from fastapi import Depends, FastAPI

app = FastAPI()


# 定义依赖类
class DepParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


# 注册路由路径
@app.get("/items/")
# 定义路径操作函数
# 参数中定义依赖项
async def read_items(params: DepParams = Depends(DepParams)):
    return params


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

上述代码的执行过程为，当Web后端服务器接收到请求后，FastAPI首先会根据请求地址匹配路径操作函数，然后会检测路径操作函数中定义的依赖项，此时FastAPI检测到依赖项params的类型是类，就会调用这个类，并创建这个类的实例，返回给路径函数。然后在路径操作函数中，执行路径操作函数中的逻辑代码(若有)，并把结果返回给调用者。

另外，在路径操作函数中，定义依赖项的代码为:

1	params: DepParams = Depends(DepParams)

其中出现了两次类名称DepParams，=后面的部分用来指定依赖类，也就是在执行代码的时候，FastAPI会实际调用的部分。=前面的部分，FastAPI不会真正使用它进行校验数据，只会用于代码提示和开发环境的自动完成功能。所以，实际上也可以这样定义：

1	params = Depends(DepParams)

这样做的结果如同使用依赖函数一样，代码可以正常工作，但失去了代码提示和开发环境的自动完成功能。

FastAPI对于上述的情况，提供了另外一种简化的方式，省略掉Depends()方法的参数部分：

1	Params: DepParams = Depends ()

这样只需要使用一次类的名称，FastAPI就会根据参数的类型定义和Depends定义共同作用，解析出所需要的依赖项。

依赖注入的嵌套

不仅是路径操作函数可以有依赖项，任何函数都可以定义依赖项，FastAPI会逐级解析并处理这些依赖关系，使软件中各个模块以低耦合的方式组装到一起。

在下文的例子中，在路径操作函数中指定了依赖函数params_extractor。同时，在这个依赖函数的参数中，又指定了依赖项query_extractor。
即，这段代码中的依赖关系有两层，第一层为主要依赖，第二层为子依赖。
当Web后端服务器接收到请求后，FastAPI首先会根据请求地址匹配到路径操作函数，然后会执行第一层的依赖项，执行第一层依赖项时检测到第二层依赖项，就会执行第二层依赖项，并将结果逐级返回，最终传递给路径操作函数。

示例代码：

from typing import Optional
import uvicorn
from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


# 定义依赖函数
def query_extractor(q: Optional[str] = None):
    return q


# 定义依赖函数
# 定义依赖项
# Cookie参数
def params_extractor(q: str = Depends(query_extractor), last_q: Optional[str] = Cookie(None)):
    if not q:
        return last_q
    return q


# 注册路由路径
@app.get("/items/")
# 定义路径操作函数
# 定义依赖项
async def read_query(params: str = Depends(params_extractor)):
    return {"params": params}


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

当程序用到的多个依赖项都依赖于某一个共同的子依赖项时，FastAPI默认会在第一次执行这个子依赖项时，将其执行结果放在缓存中，以保证对于路径操作函数的单次请求，无论定义了多少次子依赖项，这个共同的子依赖只会执行一次。

(类似于我们在《算法入门经典(Java与Python描述)：11.动态规划》所讨论的"记忆化搜索"。)

若希望每次调用子依赖项时，都能执行该子依赖项函数，而不将其执行结果放入缓存，可以在Depends()调用时，使用参数use_cache=False关闭缓存。示例代码：

1
2
3

# 使用参数关闭依赖的缓存
async def need_dependency(some_value: str = Depends(get_value, use_cache=False)):
    return ("some_value": some_ value}

在装饰器中使用依赖注入

在某些情况下，路径操作函数中并不需要依赖项的返回值，或者依赖项没有返回值，只需要在路径操作函数中调用这个依赖项，此时可以在装饰器中添加一个依赖项列表dependencies来指定需要执行的依赖项。当路径操作函数被调用时，FastAPI按顺序执行这个列表中的依赖项。

在下文的例子中，分别定义了两个依赖函数，在路径操作函数的装饰器中，使用dependencies参数设置了依赖项的列表，列表中设置的依赖项是代码中定义的两个依赖函数。这些依赖项与路径操作函数中定义的依赖项的执行方式是相同的，但是它们的执行结果不会传递给路径操作函数。如果依赖项在执行过程中抛出了异常，则会立即中止依赖的执行，也会中止路径操作函数的执行。

示例代码：

from fastapi import Depends, FastAPI, Header, Cookie, HTTPException
import uvicorn

app = FastAPI()


# 定义依赖函数
async def verify_token(x_token: str = Header(...)):
    if x_token != "my-token":
        raise HTTPException(status_code=400, detail="Token已失效")
    # 没有返回值


# 定义依赖函数
async def check_userid(userid: str = Cookie(...)):
    if userid != "9527":
        raise HTTPException(status_code=400, detail="无效的用户")
    return userid


# 装饰器中注册路由路径
# 装饰器中设置依赖项列表
@app.get("/items/", dependencies=[Depends(verify_token), Depends(check_userid)])
async def read_items():
    return "hello"


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

在构造APP时使用依赖注入

FastAPI的app应用实例也可以直接使用dependencies作为参数添加依赖项列表。

在下文的例子中，将dependencies参数放到了app应用实例中。其结果是，调用app中所有的路径操作函数时，都会执行相同的依赖项列表。

示例代码：

from fastapi import Depends, FastAPI, Header, Cookie, HTTPException
import uvicorn


# 定义依赖函数
async def verify_token(x_token: str = Header(...)):
    if x_token != "my-token":
        raise HTTPException(status_code=400, detail="Token已失效")


# 定义依赖函数
async def check_userid(userid: str = Cookie(...)):
    if userid != "9527":
        raise HTTPException(status_code=400, detail="无效的用户")
    return userid


# 设置依赖项列表
app = FastAPI(dependencies=[Depends(verify_token), Depends(check_userid)])


# 装饰器中注册路由路径
@app.get("/items/")
async def read_items():
    return "hello"


# 装饰器中注册路由路径
@app.get("/users/")
async def read_users():
    return ["张三", "李四"]


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

依赖类的可调用实例

在Python中，类本身就是"可调用"的，但如果让类的实例也成为"可调用"的，需要在类定义中实现一个特定的方法__call__()。

在下文的例子中，在依赖类的定义中实现了方法__call__()，使这个类生成的实例也是可被Depends()方法调用的。然后初始化这个类的两个实例cat和dog。在路径操作函数中实现了两个依赖项，分别是has_cat和has_dog。

示例代码：

from fastapi import Depends, FastAPI
import uvicorn

app = FastAPI()


# 定义依赖类
class PetQueryChecker:
    # 构造方法
    def __init__(self, pet_name: str):
        self.pet_name = pet_name

    # 使类的实例可调用
    def __call__(self, q: str = ""):
        if q:
            return self.pet_name in q
        return False


# 创建依赖类的可调用实例
checkcat = PetQueryChecker("cat")
# 创建依赖类的可调用实例
checkdog = PetQueryChecker("dog")


# 注册路由路径
@app.get("/pet/")
# 定义路径操作函数
async def read_query_check(
        # 依赖项，参数中有cat
        has_cat: bool = Depends(checkcat),
        # 依赖项，参数中是否有dog
        has_dog: bool = Depends(checkdog),
):
    return {"has_cat": has_cat, "has_dog": has_dog}


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

以上例子中，只定义了一个依赖类，使用不同的参数创建了两个类的实例，并加人路径操作函数的依赖项中，在路径操作函数被调用时，两个依赖项也正确返回了结果。

这就是"依赖类的可调用实例"的应用场景。想在依赖项中设置不同的参数，同时又不想再定义许多相似的函数或类。

参考资料：

Python FastAPI Web 开发从入门到项目实战 - 刘瑜安义陈逸怀喻小菲 - 华中科技大学出版社

https://fastapi.tiangolo.com/zh/tutorial/cors/

文章作者: Kaka Wan Yifan

文章链接: https://kakawanyifan.com/10906

留言板