FastAPI总结(一) 说明 在学习FastAPI的时候,我已经有了一些其他后端的使用经验,所以这篇笔记不是0基础开始的。
这部分总结分成一、二两部分:
第一部分介绍FastAPI的路径、各种参数、依赖注入等关键技术 细节。
第二部分介绍安全性 和关系型数据库 这两个重要的设计 思想;还会介绍错误处理、中间件、后台任务、多文件项目、测试等辅助技术或设计 。
大纲 第一部分
路径操作
3种重要参数
参数限定
Cookie, Header, Form, File等参数
依赖注入
第二部分
关系型数据库
安全性
多文件项目
错误处理
中间件
后台任务
其他杂项
简介 FastAPI 是一个用于构建 API 的轻量高性能web框架。
关键特性:
快速 :可与 NodeJS 和 Go 并肩的极高性能(归功于 Starlette 和 Pydantic)。
高效编码 :提高功能开发速度约 200% 至 300%。
**并发:**Django和Flask默认的服务方式都是非并发,而FastAPI则良好支持并发。
Quick Start 安装
1 2 pip install fastapi pip install "uvicorn[standard]"
示例代码
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from typing import Union 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: Union [str , None ] = None ): return {"item_id" : item_id, "q" : q}
运行
1 2 3 4 5 6 $ uvicorn main:app --reload INFO: Uvicorn running on http://127.0 .0 .1 :8000 (Press CTRL+C to quit) INFO: Started reloader process [28720 ] INFO: Started server process [28722 ] INFO: Waiting for application startup. INFO: Application startup complete.
uvicorn main:app
命令含义如下:
main
:main.py
文件(一个 Python「模块」)。
app
:在 main.py
文件中通过 app = FastAPI()
创建的对象。
--reload
:让服务器在更新代码后重新启动。仅在开发时使用该选项。
路径操作 路径操作函数就是后端中常见的,以注解形式声明的,针对一个URL路径的处理操作。
路径装饰器 就是所谓的注解,例如@app.get("/")
。显然,应该在双引号内填写这个路径操作函数针对的URL
1 2 3 4 5 6 from fastapi import FastAPI app = FastAPI()@app .get("/" ) async def root () : return {"message" : "Hello World" }
常见路径装饰器:
@app.get()
@app.post()
@app.put()
@app.delete()
以及更少见的:
@app.options()
@app.head()
@app.patch()
@app.trace()
三种参数 在 FastAPI 中,参数的类型 (是请求体参数还是查询参数)是由 参数的位置和声明方式 自动推断的。具体来说,FastAPI 会根据以下规则来判断参数的类型:
路径参数:写在URL中
1 2 3 @app.get("/items/{item_id}" ) async def read_item (item_id: int ): return {"item_id" : item_id}
查询参数:不在URL中,且非JSON格式
1 2 3 @app.get("/items/" ) async def read_items (skip: int = 0 , limit: int = 10 ): return {"skip" : skip, "limit" : limit}
请求参数:不在URL中,且为JSON格式
1 2 3 4 5 6 7 8 9 from pydantic import BaseModelclass Item (BaseModel ): name: str price: float @app.post("/items/" ) async def create_item (item: Item ): return item
后续,我们将详细讨论这3种参数的声明、限定等细节。
路径参数 即写在URL路径中的参数。
1 2 3 4 5 6 7 8 from fastapi import FastAPI app = FastAPI()@app.get("/items/{item_id}" ) async def read_item (item_id ): return {"item_id" : item_id}
顺序问题 如果有两个路径操作函数,他们有共同的前半段URL,但是在末端不一样:函数一的末端是一个单纯的路径;函数二的末端是一个路径参数。此时需要注意,将函数二声明在函数一之前,提高它的优先级。否则,当客户端访问函数一时,函数一的末端可能被视为一个路径参数。
1 2 3 4 5 6 7 8 9 10 11 12 13 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}
文件路径作为路径参数 你需要这样做:
在路径参数变量名的后面加上:path
客户端访问URL时,在{file_path:path}的部分的前面多加一个斜杠,总共2个斜杠: /files//home/johndoe/myfile.txt
1 2 3 4 5 6 7 8 from fastapi import FastAPI app = FastAPI()@app.get("/files/{file_path:path}" ) async def read_file (file_path: str ): return {"file_path" : file_path}
查询参数 路径操作函数可以接受客户端请求中位于请求体部分的数据,接收方法是将这些数据同名地声明为函数的参数,这样的参数就叫查询参数。
默认值 你还可以为这些查询参数指定默认值,当某个参数有默认值时,这个查询参数就具备可选性质。
1 2 3 4 5 6 7 8 9 10 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]
多种类型可能的参数 你可以通过|
或者Union
将某个查询参数的类型声明为多种。
1 2 3 4 5 6 7 8 9 10 from fastapi import FastAPI app = FastAPI()@app.get("/items/{item_id}" ) async def read_item (item_id: str , q: str | None = None ): if q: return {"item_id" : item_id, "q" : q} return {"item_id" : item_id}
请求参数 请求参数需要封装成类,因为他常常是多个键值对的JSON格式。这被称为请求参数模型 。
**注意:**这个封装类应当继承BaseModel
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 from fastapi import FastAPIfrom pydantic import BaseModelclass 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
查询参数模型 就像请求参数一样,查询参数也可以使用模型。然而,路径参数则几乎不使用参数模型。
注:在代码中,Field
是 Pydantic 提供的一个工具,用于为模型的字段(属性)定义额外的元数据或约束条件。它的作用是为字段提供更详细的配置,比如默认值、验证规则、描述信息等。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 from typing import Annotated, Literal from fastapi import FastAPI, Queryfrom pydantic import BaseModel, Field app = FastAPI()class FilterParams (BaseModel ): limit: int = Field(100 , gt=0 , le=100 ) offset: int = Field(0 , ge=0 ) order_by: Literal ["created_at" , "updated_at" ] = "created_at" tags: list [str ] = []@app.get("/items/" ) async def read_items (filter_query: Annotated[FilterParams, Query( )] ): return filter_query
参数限定 也称“为参数提供额外的信息和校验”。
我们通过一些特殊的函数来实现,使得接收到的参数始终处于我们想要的样子。
Query:查询参数限定 这个函数用于限定请求参数
1 2 3 4 5 6 7 8 9 10 11 12 13 from typing import Union from fastapi import FastAPI, Query app = FastAPI()@app.get("/items/" ) async def read_items (q: Union [str , None ] = Query(default=None , max_length=50 ) ): results = {"items" : [{"item_id" : "Foo" }, {"item_id" : "Bar" }]} if q: results.update({"q" : q}) return results
可选限定 注意:以下限定描述可以用于任意一种参数
default
:默认值,如果声明了默认值,则该参数为可选参数
max_length
min_length
pattern
:正则表达式
gt
:大于(g
reater t
han)
ge
:大于等于(g
reater than or e
qual)
lt
:小于(l
ess t
han)
le
:小于等于(l
ess than or e
qual)
Annotated:依赖注入 这个关键字用于依赖注入,他的语法格式为:Annotated[<type>, <ProtoData>]
在这个小节,它可以用于向一个参数注入其类型
和限定
如果单纯从语法形式上学习他的用法,可以分为两种情况:
对参数使用依赖注入时,应当将这个结果看作是类型,也就是带有限定的类型
1 2 3 4 async def read_items ( item_id: Annotated[int , Path(title="The ID of the item to get" )], q: Annotated[str | None , Query(alias="item-query" )] = None , ):
否则,把他看作值,也就是带有限定的值,但是这个限定将永远存在(尽管具体的值可能会变)
1 Username = Annotated[str , Field(min_length=3 , max_length=50 )]
Path:路径参数限定 同理
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from typing import Annotatedfrom fastapi import FastAPI, Path, Query app = FastAPI()@app.get("/items/{item_id}" ) async def read_items ( item_id: Annotated[int , Path(title="The ID of the item to get" )], q: Annotated[str | None , Query(alias="item-query" )] = None , ): results = {"item_id" : item_id} if q: results.update({"q" : q}) return results
请求参数限定 请求参数是自带限定的,他的Pydantic模型已经完成了此功能。
你还可以显式地使用Body
来限定请求参数。
下面的例子中,对importance进行了显式声明。另外,这里的显式声明是必须的,否则路径操作函数将会把importance看作查询参数。
1 2 3 4 5 6 7 8 9 @app.put("/items/{item_id}" ) async def update_item ( item_id: int , item: Item, user: User, importance: Annotated[int , Body( )] ): results = {"item_id" : item_id, "item" : item, "user" : user, "importance" : importance} return results
对于上述路径操作函数,他期望这样一个请求:
1 2 3 4 5 6 7 8 9 10 11 12 13 { "item" : { "name" : "Foo" , "description" : "The pretender" , "price" : 42.0 , "tax" : 3.2 } , "user" : { "username" : "dave" , "full_name" : "Dave Grohl" } , "importance" : 5 }
嵌入单个请求参数 如果只有一个请求参数,则不会有类似于上面”item”这样的键,如果我们需要一个这样的键,则要在Body中设置(embed=True)
。例如:
路径操作函数:
1 2 3 4 @app.put("/items/{item_id}" ) async def update_item (item_id: int , item: Annotated[Item, Body(embed=True )] ): results = {"item_id" : item_id, "item" : item} return results
期望请求:
1 2 3 4 5 6 7 8 { "item" : { "name" : "Foo" , "description" : "The pretender" , "price" : 42.0 , "tax" : 3.2 } }
而不是:
1 2 3 4 5 6 { "name" : "Foo" , "description" : "The pretender" , "price" : 42.0 , "tax" : 3.2 }
*
号的使用在 Python 中,函数参数的顺序是有规则的:没有默认值的参数(必需参数)必须放在 有默认值的参数(可选参数)之前。如果违反这个规则,Python 会报错。
而Path或者Query关键字会使得某个参数具有可选性质。当我们处理多个参数的时候,这一点让参数顺序的处理更麻烦。
FastAPI 提供了一种灵活的方式来解决这个问题:使用 *
作为函数的第一个参数。然后我们就可以无视参数顺序的规则。
参数的其他类型 以上,我们已经介绍了3种常见参数的用法,但目前这些介绍仍然局限于基本类型或者是由基本类型封装成的类模型。然而,以下这些类型也是常用的:
UUID
:
一种标准的 “通用唯一标识符” ,在许多数据库和系统中用作ID。
在请求和响应中将以 str
表示。
datetime.datetime
:
一个 Python datetime.datetime
.
在请求和响应中将表示为 ISO 8601 格式的 str
,比如: 2008-09-15T15:53:00+05:00
.
datetime.date
:
Python datetime.date
.
在请求和响应中将表示为 ISO 8601 格式的 str
,比如: 2008-09-15
.
datetime.time
:
一个 Python datetime.time
.
在请求和响应中将表示为 ISO 8601 格式的 str
,比如: 14:23:55.003
.
datetime.timedelta
:
一个 Python datetime.timedelta
.
在请求和响应中将表示为 float
代表总秒数。
Pydantic 也允许将其表示为 “ISO 8601 时间差异编码”, 查看文档了解更多信息 。
frozenset
:
在请求和响应中,作为 set
对待:
在请求中,列表将被读取,消除重复,并将其转换为一个 set
。
在响应中 set
将被转换为 list
。
产生的模式将指定那些 set
的值是唯一的 (使用 JSON 模式的 uniqueItems
)。
bytes
:
标准的 Python bytes
。
在请求和响应中被当作 str
处理。
生成的模式将指定这个 str
是 binary
“格式”。
Decimal
:
标准的 Python Decimal
。
在请求和响应中被当做 float
一样处理。
现在我们继续介绍参数。
就跟前面同理,Cookie参数和Header参数都能够:被获取、被限定、被封装为模型、作为依赖注入。
Cookie声明:Cookie()
Header声明:Header()
1 2 3 4 5 6 7 8 @app.get("/items/" ) async def read_items (ads_id: Annotated[str | None , Cookie( )] = None ): return {"ads_id" : ads_id}@app.get("/items/" ) async def read_items (user_agent: Annotated[str | None , Header( )] = None ): return {"User-Agent" : user_agent}
**注意自动转换:**Header大部分标准请求头用连字符分隔,即减号(-)。但是 user-agent 这样的变量在 Python 中是无效的。因此,默认情况下,Header 把参数名中的字符由下划线(_)改为连字符(-)来提取并存档请求头 。同时,HTTP 的请求头不区分大小写,可以使用 Python 标准样式(即 snake_case)进行声明。因此,可以像在 Python 代码中一样使用 user_agent ,无需把首字母大写为 User_Agent 等形式。如需禁用下划线自动转换为连字符,可以把 Header 的 convert_underscores 参数设置为 False。
响应模型 你可以在任意的路径操作 中使用 response_model
参数来声明用于响应的模型。
进行响应模型声明后,如果实际返回响应模型与声明响应模型相比
缺少可选字段:字段值会被设置为 None
。
缺少必选字段:会抛出 ValidationError
。
多余字段:会被自动过滤掉。
类型不匹配:FastAPI 会尝试自动转换,失败时会抛出 ValidationError
。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 class UserIn (BaseModel ): username: str password: str email: EmailStr full_name: str | None = None class UserOut (BaseModel ): username: str email: EmailStr full_name: str | None = None @app.post("/user/" , response_model=UserOut ) async def create_user (user: UserIn ) -> Any : return user
以上的例子中,返回的user
的password
将会被过滤掉。
表单参数 请求体既可以使用JSON作为数据格式,也可以使用JSON,在这一小节我们讨论前者。
Form参数数据格式示例:
1 2 3 4 POST /login/ HTTP/1.1 Content-Type : application/x-www-form-urlencoded username=john_doe&password=secret
如何让 FastAPI 识别表单数据?
使用 Form() 来声明表单字段。
确保客户端发送的请求头中 Content-Type 是 application/x-www-form-urlencoded 或 multipart/form-data(包含文件)。
JSON的 Content-Type 是 application/json
FastAPI获取Form参数示例:
1 2 3 4 5 6 7 8 from fastapi import FastAPI, Form app = FastAPI()@app.post("/login/" ) async def login (username: str = Form( ), password: str = Form( ) ): return {"username" : username}
同理,你可以想象,Form参数跟其他的参数一样,可以:被获取、被限定、被封装为模型、作为依赖注入。
文件参数 你可以使用bytes或者UploadFile获取文件参数,使用前者会读取文件到内存,使用后者会保存文件到存储。
注意 :
在请求中,文件参数的 Content-Type 是 multipart/form-data。
你可以同时请求普通表单参数和文件参数,但是一旦使用了其中任意一种,就不能请求JSON格式的参数(即上面提到的请求参数)。
1 2 3 4 5 6 7 8 9 10 11 12 13 from fastapi import FastAPI, File, UploadFile app = FastAPI()@app.post("/files/" ) async def create_file (file: bytes = File( ) ): return {"file_size" : len (file)}@app.post("/uploadfile/" ) async def create_upload_file (file: UploadFile = File( ) ): return {"filename" : file.filename}
依赖注入 一个项的部分必须依靠另一个项得到实现,另一个项就成为依赖项,声明这种关系这一动作就是依赖注入。
更通俗地说:一个函数运行时,会自动调用另一个函数。这就是依赖。
依赖注入常用于以下场景:
共享业务逻辑(复用相同的代码逻辑)
共享数据库连接
实现安全、验证、角色权限
函数依赖注入 **示例:**下面的例子展现了依赖注入在共享数据方面的作用。
read_items和read_users
会调用common_parameters
并获取相同结果。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 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
类依赖注入 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 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: CommonQueryParams = 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
你也可以这样写,这样的语法更简洁:
1 commons: CommonQueryParams = Depends()
多个依赖项 显而易见,一个依赖项也可以具有依赖项,且一个依赖项可以被使用多次。当存在一种类似于依赖路径的结构时,他们的调用顺序也是显而易见的,无需赘述。
没有返回值的依赖项 若没有返回值,你可以使用dependencies
来进行依赖注入。
1 2 3 @app.get("/items/" , dependencies=[Depends(verify_token ), Depends(verify_key )] ) async def read_items (): return [{"item" : "Foo" }, {"item" : "Bar" }]
全局依赖项 你可以通过给app注册依赖的形式,声明全局依赖项。这样一来,就可以为所有路径操作注入该依赖项。
1 app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
依赖注入配合yield 典型用法:
1 2 3 4 5 6 async def get_db (): db = DBSession() try : yield db finally : db.close()
此外,还可以配合HTTPException
使用。
yield是什么?https://fastapi.tiangolo.com/zh/tutorial/dependencies/dependencies-with-yield/#_1