Fastapi 库

官方文档：FastAPI

用户教程：教程 - 用户指南 - FastAPI

from fastapi import FastAPI
from pydantic import BaseModel
import asyncio
 
app = FastAPI()
 
class Item(BaseModel):
    name: str
    price: float
    is_offer: Union[bool, None] = None
 
 
@app.get("/")
async def read_root():
    return {"hello": "world"}
 
 
async def simulated_async_operation():
    result = await asyncio.sleep(200)  # 模拟耗时2秒的操作
    return {"message": result}
 
 
@app.get("/async-endpoint")
async def root():
    result = await simulated_async_operation()  # 在这里等待耗时操作完成
    return result
 
@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}

运行方式

运行方式：

# 根据情况将 main.py 更改为实际文件名
fastapi dev main.py --port 29997

或者通过 uvicorn 进行启动：

uvicorn main:app --reload

各个参数的属性如下：

• main：main.py 文件（一个 Python「模块」）。
• app：在 main.py 文件中通过 app = FastAPI() 创建的对象。
• —reload：让服务器在更新代码后重新启动。仅在开发时使用该选项。

API 文档

http://127.0.0.1:8000/docs - swagger 形式的文档
http://127.0.0.1:8000/redoc - redoc 形式的文档

同步与异步接口

def 代表是同步的接口，async def 代表是异步的接口。

另外关于并发 async/await 可参考：并发 async / await - FastAPI

路径参数

路径参数中是基本数据类型、或者基于 Enum 的预设值，也可以自动使用 pydantic 进行数据校验。

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}
 
 
class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"
 
@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
	pass

FastAPI 匹配路径参数时，会按顺序进行匹配，优先使用第一个匹配上的路径。例如下面 /users/me 会匹配第一个路径地址。

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

包含路径的路径参数。假设路径操作的路径为  /files/{file_path}，但需要  file_path  中也包含路径，比如，home/johndoe/myfile.txt。此时，该文件的 URL 是这样的：/files/home/johndoe/myfile.txt。

但是，OpenAPI 不支持声明包含路径的路径参数，因为这会导致测试和定义更加困难。不过，仍可使用 Starlette 内置工具在  FastAPI  中实现这一功能。

Starlette 可以采用下面的方式来配置一个包含路径的路径参数。

需要注意的是，如果要求 file_path 是 /home/johndoe/myfile.txt，那么 URL 应该是 /files//home/johndoe/myfile.txt。其中，files 和 home 之间是双斜杠。

@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

查询参数

声明的参数不是路径参数时，路径操作函数会把该参数自动解释为查询参数。如果声明的参数有默认值，那么对应的查询参数就存在默认值。

from fastapi import FastAPI
 
@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
	pass

例如，以下 URL 用于匹配上述请求：

http://127.0.0.1:8000/items/?skip=0&limit=10

同理，把默认值设为  None  即可声明可选的查询参数。FastAPI 不使用  Optional[str]  中的  Optional（只使用  str），但  Optional[str]  可以帮助编辑器发现代码中的错误。

# 3.8+
@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Union[str, None] = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}
 
# 3.10+
@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}

当路径参数和查询参数同时存在时，声明查询参数的顺序并不重要。

如果某个参数是必填写的，那么就不要声明默认值。

@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str):
	pass

请求体

FastAPI 中，请求体也是放在参数中配置的，即需要声明请求体参数：

from fastapi import FastAPI
from pydantic import BaseModel
 
class 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

请求体 + 路径参数下的示例：

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, **item.dict()}

请求体 + 路径参数 + 查询参数下的示例：

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: str | None = None):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result

函数参数按如下规则进行识别：

• 路径中声明了相同参数的参数，是路径参数
• 类型是（int、float、str、bool  等）单类型的参数，是查询参数
• 类型是  Pydantic 模型的参数，是请求体

参数的额外校验

查询参数的校验

查询参数和字符串校验 - FastAPI

from typing import Union
from fastapi import FastAPI
 
app = FastAPI()
 
 
@app.get("/items/")
async def read_items(q: Union[str, None] = None):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

添加约束条件：即使  q  是可选的，但只要提供了该参数，则该参数值不能超过 50 个字符的长度。

为此，从 FastAPI 中引入 Query 对象，显示声明为查询参数。Query(default=None)  替换默认值  None，并且可以配置其他校验机制，例如 max_length 来要求参数值的最大字符长度，pattern 来要求参数符合正则表达式。

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, min_length=3, max_length=50, pattern="^fixedquery$")
	):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

声明为必需参数

如果要求某个查询参数是必须的，但是没有没有默认值，那么只需要不声明默认参数，或者将默认值设置为 ...，或者使用 Pydantic 的 Required。

from pydantic import Required
async def read_items(q: str = Query(min_length=3)):
async def read_items(q: str = Query(default=..., min_length=3)):
async def read_items(q: str = Query(default=Required, min_length=3)):

可以声明一个参数可以接收 None 值，但它仍然是必需的。

async def read_items(q: Union[str, None] = Query(default=..., min_length=3)):

查询参数列表

async def read_items(q: Union[List[str], None] = Query(default=None)):
async def read_items(q: List[str] = Query(default=["foo", "bar"])):
async def read_items(q: list = Query(default=[])):
# http://localhost:8000/items/?q=foo&q=bar

Query 的其他属性：配置 title、description、alias 别名参数、deprecated 弃用参数。

# 别名参数，使用 item-query 来作为查询参数
# http://127.0.0.1:8000/items/?item-query=foobaritems
 
@app.get("/items/")
async def read_items(
    q: Union[str, None] = Query(
        default=None,
        title="Query string",
        description="Query string for the items",
        min_length=3,
        alias="item-query",
        deprecated=True
    ),
):
 

路径参数校验

路径参数和数值校验 - FastAPI

# 3.8+
from typing import Union
from fastapi import FastAPI, Path, Query
from typing_extensions import Annotated
 
@app.get("/items/{item_id}")
async def read_items(
    item_id: Annotated[int, Path(title="The ID of the item to get")],
    q: Annotated[Union[str, None], Query(alias="item-query")] = None,
):
 
# 3.9+
from typing import Annotated, Union
from fastapi import FastAPI, Path, Query
 
@app.get("/items/{item_id}")
async def read_items(
    item_id: Annotated[int, Path(title="The ID of the item to get")],
    q: Annotated[Union[str, None], Query(alias="item-query")] = None,
):

如果想要按需对参数排序，那么可以采取下面的方式，传递  *  作为函数的第一个参数。Python 不会对该  *  做任何事情，但是它将知道之后的所有参数都应作为关键字参数（键值对），也被称为  kwargs，来调用。即使它们没有默认值。

async def read_items(*, item_id: int = Path(title="The ID of the item to get"), q: str):

数值校验：大于等于

async def read_items(
    *, item_id: int = Path(title="The ID of the item to get", ge=1), q: str
):

数值校验：大于和小于等于

@app.get("/items/{item_id}")
async def read_items(
    *,
    item_id: int = Path(title="The ID of the item to get", gt=0, le=1000),
    q: str,
):

多个请求体参数

请求体 - 多个参数 - FastAPI

多个请求体参数下的评测方式 + 匹配示例：

class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = 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, user: User):

对应的匹配：

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    }
}

请求体中的单一值：与使用  Query  和  Path  为查询参数和路径参数定义额外数据的方式相同，FastAPI  提供了一个同等的  Body。例如，为了扩展先前的模型，你可能决定除了  item  和  user  之外，还想在同一请求体中具有另一个键  importance。

from typing import Union
from fastapi import Body, FastAPI
from pydantic import BaseModel
from typing_extensions import Annotated
 
class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = 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, user: User, importance: Annotated[int, Body(gt=0)]
):

在这种情况下，FastAPI  将期望像这样的请求体：

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    },
    "importance": 5
}

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

嵌入单个请求体参数

假设你只有一个来自 Pydantic 模型  Item  的请求体参数  item。默认情况下，FastAPI  将直接期望这样的请求体。

但是，如果你希望它期望一个拥有  item  键并在值中包含模型内容的 JSON，就像在声明额外的请求体参数时所做的那样，则可以使用一个特殊的  Body  参数  embed：

from typing import Union
 
from fastapi import Body, FastAPI
from pydantic import BaseModel
from typing_extensions import Annotated
 
app = FastAPI()
 
class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None
 
 
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
	pass

在这种情况下，FastAPI  将期望像这样的请求体：

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}

而不是：

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2
}

嵌套模型

class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None
    tags: Set[str] = set()
 
 
class Image(BaseModel):
    url: str
    name: str
 
class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None
    tags: Set[str] = set()
    image: Union[Image, None] = None
 
 
class Image(BaseModel):
    url: HttpUrl
    name: str
 
async def create_multiple_images(images: List[Image]):
 
async def create_index_weights(weights: Dict[int, float]):

请求体 - 字段 - FastAPI

请求体 - 嵌套模型 - FastAPI

Cookie 参数

@app.get("/items/")
async def read_items(ads_id: Annotated[Union[str, None], Cookie()] = None):
    return {"ads_id": ads_id}

Cookie 、Path 、Query  是兄弟类，都继承自共用的  Param  类。

Header 参数

@app.get("/items/")
async def read_items(user_agent: Annotated[Union[str, None], Header()] = None):
    return {"User-Agent": user_agent}

Header  是  Path、Query、Cookie  的兄弟类，都继承自共用的  Param  类。

Header 参数自动转换

大部分标准请求头用连字符分隔，即减号（-）。

但是  user-agent  这样的变量在 Python 中是无效的。

因此，默认情况下，Header  把参数名中的字符由下划线（_）改为连字符（-）来提取并存档请求头。

同时，HTTP 的请求头不区分大小写，可以使用 Python 标准样式（即  snake_case）进行声明。

因此，可以像在 Python 代码中一样使用  user_agent ，无需把首字母大写为  User_Agent  等形式。

如需禁用下划线自动转换为连字符，可以把  Header  的  convert_underscores  参数设置为  False

重复的请求头

使用 Python list  可以接收重复请求头所有的值。

@app.get("/items/")
async def read_items(x_token: Annotated[Union[List[str], None], Header()] = None):
    return {"X-Token values": x_token}

响应模型

在任意的路径操作中使用  response_model  参数来声明用于响应的模型。

FastAPI 使用 response_model  参数进行：将函数的返回转换为声明的类型、校验数据、生成文档。

@app.post("/items/", response_model=Item)
async def create_item(item: Item) -> Any:
 
@app.get("/items/", response_model=list[Item])
async def read_items() -> Any:

将函数的返回转换为声明的类型，可以有效的进行参数过滤。

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Union[str, None] = None
 
class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None
 
@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:
    return user

响应模型编码参数，使用 response_model_exclude_unset 属性，忽略默认值。

class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: float = 10.5
    tags: List[str] = []
 
items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}
 
@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items[item_id]

开启忽略默认值后，相应中就不存在默认值，仅有实际设置的值。如果设置的值和默认值相同，也会保留。

这么开启后，上面的接口返回的结果就和原始 items 元素的结果一致，而没有多或少的属性。

还可以使用路径操作装饰器的  response_model_include  和  response_model_exclude  参数。它们接收一个由属性名称  str  组成的  set  来包含（忽略其他的）或者排除（包含其他的）这些属性。

更多模型使用 Tip

obj.dict() 将一个 BaseModel 对象转换为 dict 形式。再加上 ** 进行 dict 解包，传入新的 BaseModel 对象中。

def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)

减少代码重复，尽量共享声明的模型。

class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None
 
class UserIn(UserBase):
    password: str
 
class UserOut(UserBase):
    pass
 
class UserInDB(UserBase):
    hashed_password: str

使用  Union  类型，即该响应可以是两种类型中的任意类型。在 OpenAPI 中可以使用  anyOf  定义。

@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem])
async def read_item(item_id: str):
	pass

使用 List 类型

@app.get("/items/", response_model=List[Item])
async def read_items():
    return items

使用 Dict 类型

@app.get("/keyword-weights/", response_model=Dict[str, float])
async def read_keyword_weights():
    return {"foo": 2.3, "bar": 3.4}

直接返回响应

直接返回响应 - FastAPI：

• 返回 Resoonse 对象，FastAPI 不会进行 Pydantic 校验，直接传递该数据。
• 使用 jsonable_encoder 函数，将无法 JSON 的数据进行转换，如 datatime、UID 等。
• Resoonse 对象可以自定义，例如 media_type="application/xml"

自定义响应 - HTML，流，文件和其他 - FastAPI：

FastAPI  默认会使用  JSONResponse  返回响应，可以通过直接返回  Response  来重载它。但如果直接返回  Response，返回数据不会自动转换，也不会自动生成文档。

还可以在  路径操作装饰器  中声明你想用的  Response。

你从  路径操作函数  中返回的内容将被放在该  Response  中。

并且如果该  Response  有一个 JSON 媒体类型（application/json），比如使用  JSONResponse  或者  UJSONResponse  的时候，返回的数据将使用你在路径操作装饰器中声明的任何 Pydantic 的  response_model  自动转换（和过滤）。

• 可用响应
  • Response
  • HTMLResponse
  • PlainTextResponse
  • JSONResponse
  • ORJSONResponse
  • UJSONResponse
  • RedirectResponse
  • StreamingResponse
    • 对类似文件的对象使用  StreamingResponse
  • FileResponse

OPENAPI 中的其他响应 - FastAPI
响应 Cookies - FastAPI
响应头 - FastAPI

响应状态码

status_code  参数接收表示 HTTP 状态码的数字。
status_code  还能接收  IntEnum  类型，比如 Python 的  http.HTTPStatus。
status_code  还可以使用  fastapi.status  中的快捷变量。

from fastapi import FastAPI, status
@app.post("/items/", status_code=201)
@app.post("/items/", status_code=status.HTTP_201_CREATED)

如果想要更换默认的状态码，则需要使用 Response 参数。

在这个临时响应对象中设置 status_code，FastAPI 将使用这个临时响应来提取状态码（也包括 cookies 和头部），并将它们放入包含你返回的值的最终响应中，该响应由任何 response_model 过滤。

@app.put("/get-or-create-task/{task_id}", status_code=200)
def get_or_create_task(task_id: str, response: Response):
    if task_id not in tasks:
        tasks[task_id] = "This didn't exist before"
        response.status_code = status.HTTP_201_CREATED
    return tasks[task_id]

额外的状态码：额外的状态码 - FastAPI

后两个有什么区别？

表单数据

请求表单

要使用表单，需预先安装  python-multipart。

@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username}

具体来说，表单数据的「媒体类型」编码一般为  application/x- www-form-urlencoded 。

但包含文件的表单编码为  multipart/form-data。文件处理详见下节。

请求文件

因为上传文件以「表单数据」形式发送。

所以接收上传文件，要预先安装  python-multipart。

File  是直接继承自  Form  的类。

from fastapi import FastAPI, File, UploadFile
 
@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):
    return {"filename": file.filename}

如果把路径操作函数参数的类型声明为  bytes，FastAPI  将以  bytes  形式读取和接收文件内容。

这种方式把文件的所有内容都存储在内存里，适用于小型文件。

UploadFile  与  bytes  相比有更多优势：

• 使用  spooled  文件：
  • 存储在内存的文件超出最大上限时，FastAPI 会把文件存入磁盘；
• 这种方式更适于处理图像、视频、二进制文件等大型文件，好处是不会占用所有内存；
• 可获取上传文件的元数据；
• 自带  file-like async  接口；
• 暴露的 Python SpooledTemporaryFile  对象，可直接传递给其他预期「file-like」对象的库。

UploadFile  支持以下  async  方法，（使用内部  SpooledTemporaryFile）可调用相应的文件方法。

• write(data)：把  data （str  或  bytes）写入文件；
• read(size)：按指定数量的字节或字符（size (int)）读取文件内容；
• seek(offset)：移动至文件  offset （int）字节处的位置；
  • 例如，await myfile.seek(0)  移动到文件开头；
  • 执行  await myfile.read()  后，需再次读取已读取内容时，这种方法特别好用；
• close()：关闭文件。

因为上述方法都是  async  方法，要搭配「await」使用。

例如，在  async 路径操作函数  内，要用以下方式读取文件内容：

contents = await myfile.read()

在普通  def 路径操作函数  内，则可以直接访问  UploadFile.file，例如：

contents = myfile.file.read()

可选文件上传

@app.post("/files/")
async def create_file(file: Union[bytes, None] = File(default=None)):
@app.post("/uploadfile/")
async def create_upload_file(file: Union[UploadFile, None] = None):

多个文件上传

@app.post("/files/")
async def create_files(files: 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.post("/files/")
async def create_file(
    file: bytes = File(), fileb: UploadFile = File(), token: str = Form()
):

处理错误

使用 HTTPException

from fastapi import FastAPI, HTTPException
 
app = FastAPI()
 
items = {"foo": "The Foo Wrestlers"}
 
 
@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

如果出错，则接收到 HTTP 状态码 - 404 及如下 JSON 响应结果：

{ "detail": "Item not found" }

添加自定义响应头

@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "There goes my error"},
        )
    return {"item": items[item_id]}

安装自定义异常处理器

可以用  @app.exception_handler()  添加自定义异常控制器

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
 
 
class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name
 
@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )
 
 
@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

覆盖请求验证异常

PlainTextResponse 只为错误返回纯文本响应，而不是返回 JSON 格式的内容。

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
 
@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="Nope! I don't like 3.")
    return {"item_id": item_id}

使用  RequestValidationError  的请求体

RequestValidationError  包含其接收到的无效数据请求的  body 。开发时，可以用这个请求体生成日志、调试错误，并返回给用户。

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
    )

复用  FastAPI  异常处理器

FastAPI 支持先对异常进行某些处理，然后再使用  FastAPI  中处理该异常的默认异常处理器。

从  fastapi.exception_handlers  中导入要复用的默认异常处理器：

from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
 
@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):
    print(f"OMG! An HTTP error!: {repr(exc)}")
    return await http_exception_handler(request, exc)
 
 
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    print(f"OMG! The client sent invalid data!: {exc}")
    return await request_validation_exception_handler(request, exc)
 
 
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}

路径操作装饰器支持的配置参数

• status_code
• tags
• summary, description
• docstring
• response_description
• deprecated

其他功能

将数据类型（如 Pydantic 模型）转换为与 JSON 兼容的数据类型（如 dict、list 等）。

如果数据库不支持 datetime 格式，需要将其转换为 JSON 兼容的数据类型，如 str。可以使用 fastapi 的 jsonable_encoder 函数进行转换。

from fastapi.encoders import jsonable_encoder
 
class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Union[str, None] = None
 
@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    fake_db[id] = json_compatible_item_data

Restful API 中使用 Patch 更新部分数据

HTTP PATCH  操作用于更新  部分  数据。即，只发送要更新的数据，其余数据保持不变。

更新部分数据时，可以在 Pydantic 模型的  .dict()  中使用  exclude_unset  参数。比如，item.dict(exclude_unset=True)。

这段代码生成的  dict  只包含创建  item  模型时显式设置的数据，而不包括默认值。

class Item(BaseModel):
    name: Union[str, None] = None
    description: Union[str, None] = None
    price: Union[float, None] = None
    tax: float = 10.5
    tags: List[str] = []
 
 
items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}
 
@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    stored_item_data = items[item_id]
    stored_item_model = Item(**stored_item_data)
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item_model.copy(update=update_data)
    items[item_id] = jsonable_encoder(updated_item)
    return updated_item

依赖注入

什么是依赖注入

编程中的 「依赖注入」 是声明代码（本文中为 路径操作函数 ）运行所需的，或要使用的「依赖」的一种方式。

然后，由系统（本文中为  FastAPI）负责执行任意需要的逻辑，为代码提供这些依赖（「注入」依赖项）。

依赖注入常用于以下场景：

• 共享业务逻辑（复用相同的代码逻辑）
• 共享数据库连接
• 实现安全、验证、角色权限
• 等……

上述场景均可以使用依赖注入，将代码重复最小化。

依赖注入示例

依赖项就是一个函数，且可以使用与路径操作函数相同的参数：

只能传给 Depends 一个参数，且该参数必须是可调用对象，比如函数。该函数接收的参数和路径操作函数的参数一样。

from fastapi import Depends, 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

通过依赖注入系统，只要告诉  FastAPI 路径操作函数  还要「依赖」其他在 路径操作函数 之前执行的内容，FastAPI  就会执行函数代码，并「注入」函数返回的结果。

其他与「依赖注入」概念相同的术语为：

• 资源（Resource）
• 提供方（Provider）
• 服务（Service）
• 可注入（Injectable）
• 组件（Component）

类作为依赖项

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
 
class CommonQueryParams:
    def __init__(self, q: Union[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)):
	# 或者 (commons: CommonQueryParams = Depends()): 也是可以的
    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

子依赖项

 
def query_extractor(q: Union[str, None] = None):
    return q
 
def query_or_cookie_extractor(
    q: str = Depends(query_extractor),
    last_query: Union[str, None] = Cookie(default=None),
):
    if not q:
        return last_query
    return q
 
@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
    return {"q_or_cookie": query_or_default}

如果在同一个 路径操作  多次声明了同一个依赖项，例如，多个依赖项共用一个子依赖项，FastAPI  在处理同一请求时，只调用一次该子依赖项。

FastAPI 不会为同一个请求多次调用同一个依赖项，而是把依赖项的返回值进行「缓存」，并把它传递给同一请求中所有需要使用该返回值的「依赖项」。

在高级使用场景中，如果不想使用「缓存」值，而是为需要在同一请求的每一步操作（多次）中都实际调用依赖项，可以把  Depends  的参数  use_cache  的值设置为  False :

async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):
    return {"fresh_value": fresh_value}

路径操作装饰器依赖项

有时，我们并不需要在路径操作函数中使用依赖项的返回值。或者说，有些依赖项不返回值，但仍要执行或解析该依赖项。

对于这种情况，不必在声明路径操作函数的参数时使用  Depends，而是可以在路径操作装饰器中添加一个由  dependencies  组成的  list。

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")
 
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.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

全局依赖项

app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])

使用 yield 依赖项

使用 yield 的依赖项 - FastAPI

安全性

中间件

使用 yield 依赖项与使用中间件的区别

@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    response = Response("Internal server error", status_code=500)
    try:
        request.state.db = SessionLocal()
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response
 
 
# Dependency
def get_db(request: Request):
    return request.state.db
 
 
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)

在此处添加中间件与 yield 的依赖项的作用效果类似，但也有一些区别：

• 中间件需要更多的代码并且更复杂一些。
• 中间件必须是一个 async 函数。
  • 如果其中有代码必须“等待”网络，它可能会在那里“阻止”您的应用程序并稍微降低性能。
  • 尽管这里的 SQLAlchemy 工作方式可能不是很成问题。
  • 但是，如果您向等待大量 I/O 的中间件添加更多代码，则可能会出现问题。
• 每个请求都会运行一个中间件。
  • 将为每个请求创建一个连接。
  • 即使处理该请求的路径操作不需要数据库。

CORS(跨域资源共享)

SQL 数据库

FastAPI + SQLAlchemy + SQLite SQL (关系型) 数据库 - FastAPI

其中关于 Pydantic 中配置 Config 类，orm_mode 将告诉 Pydantic 模型读取数据，即它不是一个 dict，而是一个 ORM 模型（或任何其他具有属性的任意对象）。

即，不仅仅尝试以 dict 形式 data['id'] 读取数据，还会尝试从属性中 data.id 中读取数据。

class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []
 
    class Config:
        orm_mode = True

汇总的效果

# sql_app/main.py
from typing import List
 
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
 
from . import crud, models, schemas
from .database import SessionLocal, engine
 
models.Base.metadata.create_all(bind=engine)
 
app = FastAPI()
 
 
# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
 
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)
 
 
@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users
 
 
@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user
 
 
@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)
 
 
@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

# sql_app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
 
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user: password@postgresserver /db"
 
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
Base = declarative_base()

APIRouter（类似 Blueprints）

from fastapi import APIRouter, Depends, HTTPException
 
from ..dependencies import get_token_header
 
router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)
 
@router.put(
    "/{item_id}",
    tags=["custom"],
    responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
    if item_id != "plumbus":
        raise HTTPException(
            status_code=403, detail="You can only update the item: plumbus"
        )
    return {"item_id": item_id, "name": "The great Plumbus"}

主程序

from fastapi import Depends, FastAPI
 
from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users
 
app = FastAPI(dependencies=[Depends(get_query_token)])
 
 
app.include_router(users.router)
app.include_router(items.router)
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)
 
 
@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

后台任务

首先导入  BackgroundTasks  并在  路径操作函数  中使用类型声明  BackgroundTasks  定义一个参数

from fastapi import BackgroundTasks, FastAPI
 
app = FastAPI()
 
 
def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)
 
 
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

使用  BackgroundTasks  也适用于依赖注入系统，你可以在多个级别声明  BackgroundTasks  类型的参数：在  路径操作函数  里，在依赖中(可依赖)，在子依赖中，等等。

FastAPI  知道在每种情况下该做什么以及如何复用同一对象，因此所有后台任务被合并在一起并且随后在后台运行：

def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)
 
def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q
 
@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

该示例中，信息会在响应发出  之后  被写到  log.txt  文件。

如果请求中有查询，它将在后台任务中写入日志。

然后另一个在  路径操作函数  生成的后台任务会使用路径参数  email  写入一条信息。

BackgroundTasks  类直接来自  starlette.background。

如果您需要执行繁重的后台计算，并且不一定需要由同一进程运行（例如，您不需要共享内存、变量等），那么使用其他更大的工具（如  Celery）可能更好。

OpenAPI 元数据和文档 URL

元数据和文档 URL - FastAPI：

• 配置应用程序元数据，如 license 等。

路径操作的高级配置 - FastAPI：

• 配置路径的 operation_id
• 将某个路径从 OpenAPI 文档中去除
• docstring 高级描述

静态文件

您可以使用  StaticFiles 从目录中自动提供静态文件。

静态文件 - FastAPI

测试

测试 - FastAPI

调试

import uvicorn
from fastapi import FastAPI
 
app = FastAPI()
 
 
@app.get("/")
def root():
    a = "a"
    b = "b" + a
    return {"hello world": b}
 
 
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

这样之后，可以打断点进行调试了。