处理请求
查询参数类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。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from typing import Optionalfrom fastapi import FastAPI, Queryimport uvicornapp = FastAPI() @app.get("/") 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个字符。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from typing import Optionalfrom fastapi import FastAPI, Queryimport uvicornapp = FastAPI() @app.get("/") 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
,是一个正则表达式。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from typing import Optionalfrom fastapi import FastAPI, Queryimport uvicornapp = FastAPI() @app.get("/") 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
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from typing import List, Optionalimport uvicornfrom fastapi import FastAPI, Queryapp = 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
,用于显示描述信息。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 from typing import Optionalimport uvicornfrom fastapi import FastAPI, Queryapp = FastAPI() @app.get("/items/") 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
。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 from typing import Optionalimport uvicornfrom fastapi import FastAPI, Queryapp = FastAPI() @app.get("/items/") 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类。
例子
添加元数据
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 import uvicornfrom fastapi import FastAPI, Pathapp = FastAPI() @app.get("/items/{item_id}") async def read_items (item_id: int = Path(..., description="项目ID是路径的一部分" ) , ) : return {"item_id" : item_id} if __name__ == '__main__' : uvicorn.run(app=app)
数值校验规则:大于、小于、等于
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 from fastapi import FastAPI, Pathimport uvicornapp = 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的值。
示例代码
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from typing import Optionalimport uvicornfrom fastapi import Cookie, FastAPIapp = FastAPI() @app.get("/items/") 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
的值。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from typing import Optionalimport uvicornfrom fastapi import FastAPI, Headerapp = FastAPI() @app.get("/items/") 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类,进行字段规则设置。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 from typing import Optionalimport uvicornfrom fastapi import Body, FastAPIfrom pydantic import BaseModel, Fieldapp = FastAPI() 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文档中显示。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 from typing import Optionalimport uvicornfrom fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI() class Item (BaseModel) : name: str description: Optional[str] = None price: float tax: Optional[float] = None class Config : schema_extra = { "example" : { "name" : "三酷猫" , "description" : "这是一个非常不错的项目" , "price" : 35.4 , "tax" : 3.2 , } } @app.put("/items/{item_id}") async def update_item (item_id: int, item: Item) : return {"item_id" : item_id, "item" : item} if __name__ == '__main__' : uvicorn.run(app=app)
当然,我们也可以使用Field类设置样例数据,示例代码:
1 2 3 4 5 6 7 class Item (BaseModel) : 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字段,记录标签列表。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 from typing import Optional, Listimport uvicornfrom fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI() class Item (BaseModel) : name: str description: Optional[str] = None price: float tax: Optional[float] = None tags: List[str] = [] @app.put("/items/{item_id}") 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来接收数据。示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 from typing import Listimport uvicornfrom fastapi import FastAPIfrom pydantic import BaseModelapp = 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来接收数据。示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from typing import Dictimport uvicornfrom fastapi import FastAPIapp = 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使用请求实例的各种属性。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from fastapi import FastAPI, Requestimport uvicornapp = FastAPI() @app.get("/host/") 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内容的响应数据。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 from fastapi import FastAPIfrom fastapi.responses import JSONResponseimport uvicornapp = FastAPI() @app.post("/cookie/") def create_cookie () : content = {"message" : "threecoolcats like cookies" } response = JSONResponse(content=content) response.set_cookie(key="user_id" , value="9527" ) return response if __name__ == '__main__' : uvicorn.run(app=app)
自定义Header数据的方式和自定义Cookie数据类似,在路径操作函数中创建响应类的实例,在创建响应类实例时,以字典的形式传入要返回的Header参数。
在下文的例子中,在创建响应类JSONResponse实例时,传入了两个参数,第一个是响应的content内容数据,第二个是响应的自定义字典型Header数据。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 from fastapi import FastAPIfrom fastapi.responses import JSONResponseimport uvicornapp = 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。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from fastapi import FastAPI, statusimport uvicornapp = 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内容。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 from fastapi import FastAPI, HTTPExceptionimport uvicornapp = FastAPI() items = {"1" : "cat" } @app.get("/items/{item_id}") async def read_item (item_id: str) : if item_id not in items: 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
将异常处理函数注册为全局异常处理器。
定义路径操作函数,在函数中主动抛出自定义异常。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 from fastapi import FastAPI, Requestfrom fastapi.responses import JSONResponseimport uvicornapp = FastAPI() 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个异常处理器,并且分别重新写了异常处理器函数的代码逻辑。在路径操作函数中,会引发两个异常,这两个异常会被重新注册的异常处理器分别捕获并处理。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 from fastapi import FastAPI, HTTPExceptionfrom fastapi.exceptions import RequestValidationErrorfrom fastapi.responses import PlainTextResponsefrom starlette.exceptions import HTTPException as StarletteHTTPExceptionimport uvicornapp = FastAPI() @app.exception_handler(StarletteHTTPException) async def http_exception_handler (request, exc) : return PlainTextResponse(str(exc.detail), status_code=exc.status_code) @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
调用内置异常处理器。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 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中的过滤器和拦截器。
在大多数情况下,FastAPI自带的中间件能够自动实现数据的接收和发送功能,无需额外处理。
但我们业可以调用中间件或自定义中间件功能,去实现某些特殊的功能。
自定义中间件
自定义中间件的过程,先定义一个中间件函数,然后在这个函数上增加装饰器@app.middleware("http")
,该函数的参数包括了请求类Request的实例request
和处理过程回调函数参数call_next
。
在下文的例子中,实现了一个完整的自定义中间件,中间件函数的内部逻辑是在接收到请求以后,记录一个时间点,等待响应结束后,再计算出处理响应的时间差,并将这个时间差记录到Header中,返回给客户端。这个操作是全局的,Web服务器端对接收到的每个请求都会调用中间件执行操作。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 import timeimport uvicornfrom fastapi import FastAPI, Requestapp = FastAPI() @app.middleware("http") async def add_process_time_header (request: Request, call_next) : start_time = time.time() response = await call_next(request) process_time = time.time() - start_time 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中间件可用的配置项。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 from fastapi import FastAPIfrom fastapi.middleware.cors import CORSMiddlewareimport uvicornapp = FastAPI() origins = [ "http://kakawanyifan.com" , "https://kakawanyifan.com" , "http://localhost" , "http://localhost:8080" , ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True , allow_methods=["*" ], 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()
方法,将该中间件添加到应用上。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 from fastapi import FastAPIfrom fastapi.middleware.httpsredirect import HTTPSRedirectMiddlewareimport uvicornapp = 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响应,表示资源不可用。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 from fastapi import FastAPIfrom fastapi.middleware.trustedhost import TrustedHostMiddlewareimport uvicornapp = 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压缩,否则将会传递原始数据。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 from fastapi import FastAPIfrom fastapi.middleware.gzip import GZipMiddlewareimport uvicornapp = 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中使用依赖注入的方式。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 from typing import Optionalimport uvicornfrom fastapi import Depends, FastAPIapp = 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
。
从依赖函数中获取结果。
将该结果返回给路径操作函数中的定义的返回数据对象。
使用类实现依赖注入
除了通过依赖函数的方式实现依赖注人,也可以通过依赖类完成相同的依赖注入功能。
在下文的例子中,定义了依赖类,没有使用依赖函数,将多个参数封装成一个参数类,不但增加了代码的可读性,还可以在开发工具上使用代码提示和开发环境的自动完成等辅助功能。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 from typing import Optionalimport uvicornfrom fastapi import Depends, FastAPIapp = 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首先会根据请求地址匹配到路径操作函数,然后会执行第一层的依赖项,执行第一层依赖项时检测到第二层依赖项,就会执行第二层依赖项,并将结果逐级返回,最终传递给路径操作函数。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 from typing import Optionalimport uvicornfrom fastapi import Cookie, Depends, FastAPIapp = FastAPI() def query_extractor (q: Optional[str] = None) : return q 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
参数设置了依赖项的列表,列表中设置的依赖项是代码中定义的两个依赖函数。这些依赖项与路径操作函数中定义的依赖项的执行方式是相同的,但是它们的执行结果不会传递给路径操作函数。如果依赖项在执行过程中抛出了异常,则会立即中止依赖的执行,也会中止路径操作函数的执行。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 from fastapi import Depends, FastAPI, Header, Cookie, HTTPExceptionimport uvicornapp = 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中所有的路径操作函数时,都会执行相同的依赖项列表。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 from fastapi import Depends, FastAPI, Header, Cookie, HTTPExceptionimport uvicornasync 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
。
示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 from fastapi import Depends, FastAPIimport uvicornapp = 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 ( has_cat: bool = Depends(checkcat) , has_dog: bool = Depends(checkdog) , ) : return {"has_cat" : has_cat, "has_dog" : has_dog} if __name__ == '__main__' : uvicorn.run(app)
以上例子中,只定义了一个依赖类,使用不同的参数创建了两个类的实例,并加人路径操作函数的依赖项中,在路径操作函数被调用时,两个依赖项也正确返回了结果。
这就是"依赖类的可调用实例"的应用场景。想在依赖项中设置不同的参数,同时又不想再定义许多相似的函数或类。