学习 py 后端,flask 太老,django 太重,就 FastAPI 了
Ref: Learn -
FastAPI
Quick Setup
基于 uv
1 2 3 uv init awesome-project --bare cd awesome-projectuv add "fastapi[standard]"
编辑 main.py
main.py 1 2 3 4 5 6 7 8 from fastapi import FastAPIapp = FastAPI() @app.get("/" ) async def root (): return {"message" : "Hello World" }
运行项目,可以先配置 toml
pyproject.toml 1 2 [tool.fastapi] entrypoint = "main:app"
然后
随后可以在 /docs 和 /redoc 查看自动生成的
API 文档。
Path Parameters
类似大多数路由框架,FastAPI 也支持路径参数。
1 2 3 @app.get("/items/{item_id}" ) async def read_item (item_id ): return {"item_id" : item_id}
还可以标注路径参数的类型,且当配置了类型时,框架会自动进行类型校验以及转换,例如从
str 到 int
1 2 3 @app.get("/items/{item_id}" ) async def read_item (item_id: int ): return {"item_id" : item_id}
路由顺序
例如你有两个路由:
1 2 3 4 5 6 7 8 @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}
这个情况下必须使 /users/me 的路由在
/users/{user_id} 之前,否则 /users/me
会被匹配到
/users/{user_id},从而导致无法访问,同样的对于两个路径相同的路由,后一个会被前一个覆盖。
枚举类型
通过枚举类型可以限定路径参数的值,例如:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 from enum import Enumfrom fastapi import FastAPIclass 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 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" }
Path
Parameters and Qeury Parameters at the same time
与 Query 类似
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from typing import Annotatedfrom fastapi import FastAPI, Path, Queryapp = 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
这里有一些细节,如果你不需要定义 query
的默认值,但需要给 path 添加默认值/验证规则,要么你使用
Annotated 以类型形式定义,要么你得把 path
之后的参数全都加上默认值,这里有一个 trick,可以用 *
放在签名的开始,让后续所有参数都转为 kwargs
于是就无需关注有无默认值的顺序。
Number validations
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from typing import Annotatedfrom fastapi import FastAPI, Pathapp = FastAPI() @app.get("/items/{item_id}" ) async def read_items ( item_id: Annotated[int , Path(title="The ID of the item to get" , ge=1 )], q: str ): results = {"item_id" : item_id} if q: results.update({"q" : q}) return results
与 LaTeX 语法类似
ge - greater than or equal to
gt - greater than
le - less than or equal to
lt - less than
Query Parameters
任何非路径参数的参数都可以被认为是查询参数,例如:
1 2 3 4 5 6 7 8 9 10 from fastapi import FastAPIapp = 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]
同样的,如果你在参数上标注了类型,那么框架也会自动进行类型校验和转换,包括
= 的默认值
Optional parameters
1 2 3 4 5 6 7 8 9 10 from fastapi import FastAPIapp = 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}
通过 None
的默认值可以声明可选参数,反之若无默认值则为必选参数
Query Parameter Models
可以使用 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, Fieldapp = 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 Parameters and
String Validations
1 2 3 4 5 6 7 8 9 10 11 12 13 from typing import Annotatedfrom fastapi import FastAPI, Queryapp = FastAPI() @app.get("/items/" ) async def read_items (q: Annotated[str | None , Query(max_length=50 )] = None ): results = {"items" : [{"item_id" : "Foo" }, {"item_id" : "Bar" }]} if q: results.update({"q" : q}) return results
采用 Annotated
来声明查询参数的复合类型(验证规则),例如上例中声明了 q
的最大长度为 50。
正则
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 from typing import Annotatedfrom fastapi import FastAPI, Queryapp = FastAPI() @app.get("/items/" ) async def read_items ( q: Annotated[ str | None , Query(min_length=3 , max_length=50 , pattern="^fixedquery$" ) ] = None , ): results = {"items" : [{"item_id" : "Foo" }, {"item_id" : "Bar" }]} if q: results.update({"q" : q}) return results
默认值
1 2 3 4 5 6 7 8 9 10 11 12 13 from typing import Annotatedfrom fastapi import FastAPI, Queryapp = FastAPI() @app.get("/items/" ) async def read_items (q: Annotated[str , Query(min_length=3 )] = "fixedquery" ): results = {"items" : [{"item_id" : "Foo" }, {"item_id" : "Bar" }]} if q: results.update({"q" : q}) return results
列表查询参数
1 2 3 4 5 6 7 8 9 10 11 from typing import Annotatedfrom fastapi import FastAPI, Queryapp = FastAPI() @app.get("/items/" ) async def read_items (q: Annotated[list [str ], Query( )] = ["foo" , "bar" ] ): query_items = {"q" : q} return query_items
你必须用 Annotated 以及 Query
来显式声明列表类型的查询参数,否则会被解析为请求体。
查询参数别名
1 2 3 4 5 6 7 8 9 10 11 12 13 from typing import Annotatedfrom fastapi import FastAPI, Queryapp = FastAPI() @app.get("/items/" ) async def read_items (q: Annotated[str | None , Query(alias="item-query" )] = None ): results = {"items" : [{"item_id" : "Foo" }, {"item_id" : "Bar" }]} if q: results.update({"q" : q}) return results
对于类似 item-query 这样的查询参数,Python
中的变量名不能包含 -,所以需要通过 alias
来指定查询参数的别名。
这个 alias 不能理解为别名,因为这个情况下无法使用 q
传参
自定义校验
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 import randomfrom typing import Annotatedfrom fastapi import FastAPIfrom pydantic import AfterValidatorapp = FastAPI() data = { "isbn-9781529046137" : "The Hitchhiker's Guide to the Galaxy" , "imdb-tt0371724" : "The Hitchhiker's Guide to the Galaxy" , "isbn-9781439512982" : "Isaac Asimov: The Complete Stories, Vol. 2" , } def check_valid_id (id : str ): if not id .startswith(("isbn-" , "imdb-" )): raise ValueError('Invalid ID format, it must start with "isbn-" or "imdb-"' ) return id @app.get("/items/" ) async def read_items ( id : Annotated[str | None , AfterValidator(check_valid_id )] = None , ): if id : item = data.get(id ) else : id , item = random.choice(list (data.items())) return {"id" : id , "name" : item}
这里 AfterValidator 可以在参数解析后(str
校验通过)进行自定义校验,若校验失败则会返回 422 错误。
如果需要在检查时使用外部组件,例如数据库,HTTP 请求,则需要使用
FastAPI 的依赖注入
Request Body
通过 Pydantic 来声明请求体
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
路由函数参数最终会按如下规则解析:
如果一个参数它在路径中,那么它会被解析为路径参数
如果一个参数是一个简单类型(str、int、float、bool),那么它会被解析为查询参数
如果一个参数是一个 Pydantic
模型类的实例,那么它会被解析为请求体
单一参数
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 from typing import Annotatedfrom fastapi import FastAPI, Pathfrom pydantic import BaseModelapp = FastAPI() class Item (BaseModel ): name: str description: str | None = None price: float tax: float | 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=0 , le=1000 )], q: str | None = None , item: Item | None = None , ): results = {"item_id" : item_id} if q: results.update({"q" : q}) if item: results.update({"item" : item}) return results
上面这个 example 期望一个如下形式的 JSON item
1 2 3 4 5 6 { "name" : "Foo" , "description" : "The pretender" , "price" : 42.0 , "tax" : 3.2 }
多个参数
当你有多个 Body Parameters 时
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 from fastapi import FastAPIfrom pydantic import BaseModelapp = 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: int , item: Item, user: User ): results = {"item_id" : item_id, "item" : item, "user" : user} return results
这个例子期望如下形式的 JSON
1 2 3 4 5 6 7 8 9 10 11 12 { "item" : { "name" : "Foo" , "description" : "The pretender" , "price" : 42.0 , "tax" : 3.2 } , "user" : { "username" : "dave" , "full_name" : "Dave Grohl" } }
通过 key 来区分不同的 Body Parameters,并根据变量名作为 key 从 JSON
中提取
基本类型的 Body
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 Annotatedfrom fastapi import Body, FastAPIfrom pydantic import BaseModelapp = 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: int , item: Item, user: User, importance: Annotated[int , Body( )] ): results = {"item_id" : item_id, "item" : item, "user" : user, "importance" : importance} return results
需要强制标识 Body(),与
Query()、Path()、Header()、Cookie()
一样,来告诉框架这是一个请求体参数,而不是查询参数或路径参数。
总结来说:
参数名出现在路径中 → Path
参数是简单类型(int、str、float 等) → Query
参数是 Pydantic 模型 → Body
Fields
如果你需要参数校验,那么你需要用 Pydantic 模型进行而不是 FastAPI
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 from typing import Annotatedfrom fastapi import Body, FastAPIfrom pydantic import BaseModel, Fieldapp = FastAPI() class Item (BaseModel ): name: str description: str | None = Field( default=None , title="The description of the item" , max_length=300 ) price: float = Field(gt=0 , description="The price must be greater than zero" ) tax: float | None = None @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
Nested Models
直接用 Pydantic 的嵌套模型即可
Cookie Parameters
同样类似
1 2 3 4 5 6 7 8 9 10 from typing import Annotatedfrom fastapi import Cookie, FastAPIapp = FastAPI() @app.get("/items/" ) async def read_items (ads_id: Annotated[str | None , Cookie( )] = None ): return {"ads_id" : ads_id}
Cookie Parameter Models
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 from typing import Annotatedfrom fastapi import Cookie, FastAPIfrom pydantic import BaseModelapp = 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
类似处理
1 2 3 4 5 6 7 8 9 10 from typing import Annotatedfrom fastapi import FastAPI, Headerapp = FastAPI() @app.get("/items/" ) async def read_items (user_agent: Annotated[str | None , Header( )] = None ): return {"User-Agent" : user_agent}
Automatic conversion
即连字符的处理,例如 User-Agent 会被默认转换为
user_agent,也可以关闭
1 2 3 4 5 6 7 8 9 10 11 12 from typing import Annotatedfrom fastapi import FastAPI, Headerapp = FastAPI() @app.get("/items/" ) async def read_items ( strange_header: Annotated[str | None , Header(convert_underscores=False )] = None , ): return {"strange_header" : strange_header}
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 from typing import Annotatedfrom fastapi import FastAPI, Headerfrom pydantic import BaseModelapp = FastAPI() class CommonHeaders (BaseModel ): 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
Response Model - Return Type
通过用 Pydantic 模型标注返回类型,FastAPI 会自动将返回值转换为
JSON
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 fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI() class Item (BaseModel ): name: str description: str | None = None price: float tax: float | None = None tags: list [str ] = [] @app.post("/items/" ) async def create_item (item: Item ) -> Item: return item @app.get("/items/" ) async def read_items () -> list [Item]: return [ Item(name="Portal Gun" , price=42.0 ), Item(name="Plumbus" , price=32.0 ), ]
response_model Parameter
response_model 是用来解决 return type 和 API 的 Output schema
不一致的问题,主要是转为一个更小属性类时做自动 filter
或者说转换用,比如下面这个例子,从 UserIn 转为
UserOut,从而隐藏了 password 字段。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 from typing import Any from fastapi import FastAPIfrom pydantic import BaseModel, EmailStrapp = FastAPI() 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
不过自然的,也就可以用基类来实现
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 from fastapi import FastAPIfrom pydantic import BaseModel, EmailStrapp = FastAPI() class BaseUser (BaseModel ): username: str email: EmailStr full_name: str | None = None class UserIn (BaseUser ): password: str @app.post("/user/" ) async def create_user (user: UserIn ) -> BaseUser: return user
Return a Response Directly
1 2 3 4 5 6 7 8 9 10 11 from fastapi import FastAPI, Responsefrom fastapi.responses import JSONResponse, RedirectResponseapp = FastAPI() @app.get("/portal" ) async def get_portal (teleport: bool = False ) -> Response: if teleport: return RedirectResponse(url="https://www.youtube.com/watch?v=dQw4w9WgXcQ" ) return JSONResponse(content={"message" : "Here's your interdimensional portal." })
Response Status Code
这里是直接设置一个路由的默认返回状态码
1 2 3 4 5 6 7 8 from fastapi import FastAPIapp = FastAPI() @app.post("/items/" , status_code=201 ) async def create_item (name: str ): return {"name" : name}
或者
1 2 3 4 5 6 7 8 from fastapi import FastAPI, statusapp = FastAPI() @app.post("/items/" , status_code=status.HTTP_201_CREATED ) async def create_item (name: str ): return {"name" : name}
To use forms, first install python-multipart.
$ uv add python-multipart
1 2 3 4 5 6 7 8 9 10 from typing import Annotatedfrom fastapi import FastAPI, Formapp = FastAPI() @app.post("/login/" ) async def login (username: Annotated[str , Form( )], password: Annotated[str , Form( )] ): return {"username" : username}
Form is a class that inherits directly from Body.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from typing import Annotatedfrom fastapi import FastAPI, Formfrom pydantic import BaseModelapp = FastAPI() class FormData (BaseModel ): username: str password: str @app.post("/login/" ) async def login (data: Annotated[FormData, Form( )] ): return data
Request Files
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from typing import Annotatedfrom fastapi import FastAPI, File, UploadFileapp = FastAPI() @app.post("/files/" ) async def create_file (file: Annotated[bytes , File( )] ): return {"file_size" : len (file)} @app.post("/uploadfile/" ) async def create_upload_file (file: UploadFile ): return {"filename" : file.filename}
通过 File() 来声明一个文件参数,会以
Form Data 的方式处理,这里声明了类型为 bytes
会以字节数组形式给出,这会使其存储与内存中,不适合打文件
而 UploadFile
是一个更高级的文件类型,支持异步读取、文件名、内容类型等,同时其采用了
spooling。
同样也可以做可选上传
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 from typing import Annotatedfrom fastapi import FastAPI, File, UploadFileapp = FastAPI() @app.post("/files/" ) async def create_file (file: Annotated[bytes | None , File( )] = 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}
Multiple File Uploads
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 from typing import Annotatedfrom fastapi import FastAPI, File, UploadFilefrom fastapi.responses import HTMLResponseapp = FastAPI() @app.post("/files/" ) async def create_files (files: Annotated[list [bytes ], File( )] ): return {"file_sizes" : [len (file) for file in files]} @app.post("/uploadfiles/" ) async def create_upload_files (files: list [UploadFile] ): return {"filenames" : [file.filename for file in files]} @app.get("/" ) async def main (): content = """ <body> <form action="/files/" enctype="multipart/form-data" method="post"> <input name="files" type="file" multiple> <input type="submit"> </form> <form action="/uploadfiles/" enctype="multipart/form-data" method="post"> <input name="files" type="file" multiple> <input type="submit"> </form> </body> """ return HTMLResponse(content=content)
这里的 List[bytes] 说明是多个文件,而 bytes
是单个文件,因为默认以 List 包装