FastAPI教程II

本文参考FastAPI教程https://fastapi.tiangolo.com/zh/tutorial

Cookie参数

定义Cookie参数与定义QueryPath参数一样。
具体步骤如下:

  • 导入Cookiefrom fastapi import Cookie
  • 声明Cookie参数,声明Cookie参数的方式与声明QueryPath参数相同,第一个值是默认值,还可以传递所有验证参数和注释参数
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声明cookie参数,否则该参数会被解释成查询参数。

测试代码如下:

import requests# 设置你的 FastAPI 服务地址
base_url = "http://localhost:8000"# 发送带有 Cookie 的 GET 请求
def test_read_items_with_cookie():# 设置 Cookie 值cookies = {"ads_id": "your_ads_id_here"}# 发送 GET 请求,带上 Cookieresponse = requests.get(f"{base_url}/items/", cookies=cookies)# 打印响应内容print(response.json())# 执行测试
if __name__ == "__main__":test_read_items_with_cookie()

得到如下结果:

{'ads_id': 'your_ads_id_here'}

Header参数

定义Header参数的方式与定义QueryPathCookie参数相同。

步骤如下:

  • 导入Header
  • 声明Header参数:使用和PathQueryCookie一样的结构定义header参数,第一个值是默认值,还可以传递所有验证参数或注释参数
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}
  • 必须使用Header声明header参数,否则该参数会被解释成查询参数。

自动转换

HeaderPathQueryCookie提供了更多功能。

大部分标准请求头用连字符分隔,即减号(-)

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

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

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

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

如需禁用下划线自动转换为连字符,可以把Headerconvert_underscores参数设置为False

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}

注意,使用 convert_underscores = False 要慎重,有些 HTTP 代理和服务器不支持使用带有下划线的请求头。

重复的请求头

有时,可能需要接收重复的请求头。即同一个请求头有多个值。

类型声明中可以使用list定义多个请求头。

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

例如,声明X-Token多次出现的请求头,可以写成这样:

from typing import Annotatedfrom fastapi import FastAPI, Headerapp = FastAPI()@app.get("/items/")
async def read_items(x_token: Annotated[list[str] | None, Header()] = None):return {"X-Token values": x_token}

与路径操作通信时,以下面的方式发送两个 HTTP 请求头:

X-Token: foo
X-Token: bar

响应结果是:

{"X-Token values": ["bar","foo"]
}

这部分有疑问,怎么发送两个同样的请求头,如果直接发送字典形式,那么同名的会覆盖。

响应模型

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

  • @app.get()
  • @app.post()
  • @app.put()
  • @app.delete()
  • 等等
from typing import Anyfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonetags: list[str] = []@app.post("/items/", response_model=Item)
async def create_item(item: Item) -> Any:return item@app.get("/items/", response_model=list[Item])
async def read_items() -> Any:return [{"name": "Portal Gun", "price": 42.0},{"name": "Plumbus", "price": 32.0},]

注意,response_model是【装饰器】方法(getpost等)的一个参数。不像之前的所有参数和请求体,它不属于路径操作函数

它接收的类型与你将为Pydantic模型属性所声明的类型相同,因此它可以是一个Pydantic模型,但也可以是一个由Pydantic模型组成的list,例如List[Item]

FastAPI将使用此response_model来:

  • 将输出数据转换为其声明的类型
  • 校验数据
  • 在OpenAPI的路径操作中为响应添加一个JSON Schema
  • 并在自动生成文档系统中使用

但最重要的是:

  • 会将输出数据限制在该模型定义内。

返回与输入相同的数据

现在我们声明一个UserIn模型,它将包含一个明文密码属性。

我们正在使用此模型声明输入数据,并使用同一模型声明输出数据:

from typing import Unionfrom fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp = FastAPI()class UserIn(BaseModel):username: strpassword: stremail: EmailStrfull_name: Union[str, None] = None# Don't do this in production!
@app.post("/user/")
async def create_user(user: UserIn) -> UserIn:return user

现在,每当浏览器使用一个密码创建用户时,API都会在响应中返回相同的密码。

在这个案例中,这可能不算是问题,因为用户自己正在发送密码。

但是,如果我们在其他的路径操作中使用相同的模型,则可能会将用户的密码发送给每个客户端。

永远不要存储用户的明文密码,也不要在响应中发送密码。

添加输出模型

相反,我们可以创建一个有明文密码的输入模型和一个没有明文密码的输出模型:

from typing import Anyfrom fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp = FastAPI()class UserIn(BaseModel):username: strpassword: stremail: EmailStrfull_name: str | None = Noneclass UserOut(BaseModel):username: stremail: EmailStrfull_name: str | None = None@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:return user

这样,即使我们的路径操作函数将会返回包含密码的相同输入用户,我们已经将response_model声明为了不包含密码的UserOut模型,因此,FastAPI将会负责过滤掉未在输出模型中声明的所有数据(使用Pydantic)。

在文档中查看

当你查看自动化文档时,你可以检查输入模型和输出模型是否都具有自己的JSON Schema,并且两种模型都将在交互式API文档中使用。

响应模型编码参数

你的响应模型可以具有默认值,例如:

from typing import List, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: float = 10.5tags: 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]
  • description: Union[str, None] = None 具有默认值 None
  • tax: float = 10.5 具有默认值 10.5.
  • tags: List[str] = [] 具有一个空列表作为默认值: [].

但如果它们并没有存储实际的值,你可能想从结果中忽略它们的默认值。

举个例子,当你在NoSQL数据库中保存了许多具有可选属性的模型,但你又不想发送充满默认值的很长的JSON响应。

使用response_model_exclude_unset参数

你可以设置路径操作装饰器的response_model_exclude_unset=True参数:

from typing import List, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: float = 10.5tags: 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]

然后响应中将不会包含那些默认值,而是仅有实际设置的值。

因此,如果你向路径操作发送ID为Foo的商品的请求,则响应(不包括默认值)将为:

{"name": "Foo","price": 50.2
}

你还可以使用:

  • response_model_exclude_defaults=True
  • response_model_exclude_none=True
    参考 Pydantic 文档 中对 exclude_defaultsexclude_none 的描述。

response_model_includeresponse_model_exclude

你还可以使用路径操作装饰器的 response_model_includeresponse_model_exclude 参数。

它们接收一个由属性名称 str 组成的 set来包含(忽略其他的)或者排除(包含其他的)这些属性。

如果你只有一个 Pydantic 模型,并且想要从输出中移除一些数据,则可以使用这种快捷方法。
在这里插入图片描述

from typing import Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: float = 10.5items = {"foo": {"name": "Foo", "price": 50.2},"bar": {"name": "Bar", "description": "The Bar fighters", "price": 62, "tax": 20.2},"baz": {"name": "Baz","description": "There goes my baz","price": 50.2,"tax": 10.5,},
}@app.get("/items/{item_id}/name",response_model=Item,response_model_include={"name", "description"},
)
async def read_item_name(item_id: str):return items[item_id]@app.get("/items/{item_id}/public", response_model=Item, response_model_exclude={"tax"})
async def read_item_public_data(item_id: str):return items[item_id]

使用list而不是set

如果你忘记使用 set 而是使用 listtuple,FastAPI 仍会将其转换为 set 并且正常工作:

from typing import Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: float = 10.5items = {"foo": {"name": "Foo", "price": 50.2},"bar": {"name": "Bar", "description": "The Bar fighters", "price": 62, "tax": 20.2},"baz": {"name": "Baz","description": "There goes my baz","price": 50.2,"tax": 10.5,},
}@app.get("/items/{item_id}/name",response_model=Item,response_model_include=["name", "description"],
)
async def read_item_name(item_id: str):return items[item_id]@app.get("/items/{item_id}/public", response_model=Item, response_model_exclude=["tax"])
async def read_item_public_data(item_id: str):return items[item_id]

更多模型

多个关联模型这种情况很常见。

特别是用户模型,因为:

  • 输入模型应该含密码
  • 输出模型不应含密码
  • 数据库模型需要加密的密码

**千万不要存储用户的明文密码。始终存储可以进行验证的安全哈希值。

如果不了解这方面的知识,请参阅安全性中的章节,了解什么是密码哈希。**

多个模型

下面的代码展示了不同模型处理密码字段的方式,及使用位置的大致思路:

from fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp = FastAPI()class UserIn(BaseModel):username: strpassword: stremail: EmailStrfull_name: str | None = Noneclass UserOut(BaseModel):username: stremail: EmailStrfull_name: str | None = Noneclass UserInDB(BaseModel):username: strhashed_password: stremail: EmailStrfull_name: str | None = Nonedef fake_password_hasher(raw_password: str):return "supersecret" + raw_passworddef 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)print("User saved! ..not really")return user_in_db@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):user_saved = fake_save_user(user_in)return user_saved

**user_in_dict()简介

Pydantic的.dict()

user_in是类UserIn的Pydantic模型。

Pydantic模型支持.dict方法,能返回包含模型数据的字典。

因此,如果使用如下方式创建Pydantic对象user_in

user_in = UserIn(username="john", password="secret", email="john.doe@example.com")

就能以如下方式调用:

user_dict = user_in.dict()

现在,变量user_dict中的就是包含数据的字典(变量user_dict是字典,不是Pydantic模型对象)。

以如下方式调用:

print(user_dict)

输出的就是 Python 字典:

{'username': 'john','password': 'secret','email': 'john.doe@example.com','full_name': None,
}

解包dict

把字典user_dict**user_dict形式传递给函数(或类),Python会执行解包操作。它会把user_dict的键和值作为关键字参数直接传递。

因此,接着上面的user_dict继续编写如下代码:

UserInDB(**user_dict)

就会生成如下结果:

UserInDB(username="john",password="secret",email="john.doe@example.com",full_name=None,
)

或更精准,直接把可能会用到的内容与user_dict一起使用:

UserInDB(username = user_dict["username"],password = user_dict["password"],email = user_dict["email"],full_name = user_dict["full_name"],
)

用其它模型中的内容生成Pydantic模型

上例中,从user_in.dict()中得到了user_dict,下面的代码:

user_dict = user_in.dict()
UserInDB(**user_dict)

等效于:

UserInDB(**user_in.dict())

因为 user_in.dict() 是字典,在传递给 UserInDB 时,把 ** 加在 user_in.dict() 前,可以让 Python 进行解包。

这样,就可以用其它Pydantic模型中的数据生成Pydantic模型。

解包dict和更多关键字

接下来,继续添加关键字参数hashed_password=hashed_password,例如:

UserInDB(**user_in.dict(), hashed_password=hashed_password)

输出结果如下:

UserInDB(username = user_dict["username"],password = user_dict["password"],email = user_dict["email"],full_name = user_dict["full_name"],hashed_password = hashed_password,
)

减少重复

FastAPI的核心思想就是减少代码重复。

代码重复会导致bug、安全问题、代码失步等问题(更新了某个位置的代码,但没有同步更新其它位置的代码)。

上面的这些模型共享了大量数据,拥有重复的属性名和类型。

FastAPI可以做到更好。

声明UserBase模型作为其它模型的基类,然后,用该类衍生出继承其属性(类型声明、验证等)的子类。

所有数据转换、校验、文档等功能仍将正常运行。

这样,就可以仅声明模型之间的差异部分(具有明文的password、具有hashed_password以及不包括密码)。

通过这种方式,可以只声明模型之间的区别(分别包含明文密码、哈希密码,以及无密码的模型)。

from fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp = FastAPI()class UserBase(BaseModel):username: stremail: EmailStrfull_name: str | None = Noneclass UserIn(UserBase):password: strclass UserOut(UserBase):passclass UserInDB(UserBase):hashed_password: strdef fake_password_hasher(raw_password: str):return "supersecret" + raw_passworddef 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)print("User saved! ..not really")return user_in_db@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):user_saved = fake_save_user(user_in)return user_saved

Union或者anyOf

响应可以声明为两种类型的Union类型,即该响应可以是两种类型中的任意类型。

在OpenAPI中可以使用anyOf定义

为此,请使用Python标准类型提示typing.Union

from typing import Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class BaseItem(BaseModel):description: strtype: strclass CarItem(BaseItem):type: str = "car"class PlaneItem(BaseItem):type: str = "plane"size: intitems = {"item1": {"description": "All my friends drive a low rider", "type": "car"},"item2": {"description": "Music is my aeroplane, it's my aeroplane","type": "plane","size": 5,},
}@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem])
async def read_item(item_id: str):return items[item_id]

模型列表

使用同样的方式也可以声明由对象列表构成的响应。

为此,请使用标准的Python typing.List

from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: stritems = [{"name": "Foo", "description": "There comes my hero"},{"name": "Red", "description": "It's my aeroplane"},
]@app.get("/items/", response_model=list[Item])
async def read_items():return items

任意dict构成的响应

任意的dict都能用于声明响应,只要声明键和值的类型,无需使用Pydantic模型。

事先不知道可用的字段/属性名时(Pydantic模型必须知道字段是什么),这种方式特别有用。

此时,可以使用typing.Dict

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

响应状态码

与指定响应模型的方式相同,在以下任意路径操作中,可以使用status_code参数声明用于响应的HTTP状态码:

  • @app.get()
  • @app.post()
  • @app.put()
  • @app.delete()
  • 等……
from fastapi import FastAPIapp = FastAPI()@app.post("/items/", status_code=201)
async def create_item(name: str):return {"name": name}

status_code参数接收表示HTTP状态码的数字。

它可以:

  • 在响应中返回状态码
  • 在OpenAPI概图(及用户界面)中存档

关于HTTP状态码

在HTTP协议中,发送3位数的数字状态码是相应的一部分。

这些状态码都具有便于识别的关联名称,但是重要的还是数字。

简言之:

  • 100及以上的状态码用于返回信息。这类状态码很少直接使用。具有这些状态码的响应不能包含响应体。
  • 200及以上的状态码用于表示成功。这些状态码是最常用的
    • 200是默认状态代码,表示一切正常
    • 201表示已创建,通常在数据库中创建新纪录后使用
    • 204是一种特殊的例子,表示无内容。该响应在没有为客户端返回内容时使用,因此,该响应不能包含响应体
  • 300及以上的状态码用于重定向。具有这些状态码的响应不一定包含响应体,但304未修改是个例外,该响应不得包含响应体。
  • 400及以上的状态码用于表示客户端错误。这些可能是第二常用的类型
    • 404,用于未找到响应
    • 对于来自客户端的一般错误,可以只使用400
  • 500及以上的状态码用于表示服务器端错误。几乎永远不会直接使用这些状态码。应用代码或服务器出现问题时,会自动返回这些状态代码

状态码及适用场景的详情,请参阅MDN 的 HTTP 状态码文档。

状态码名称快捷方式

再看下之前的例子:

from fastapi import FastAPIapp = FastAPI()@app.post("/items/", status_code=201)
async def create_item(name: str):return {"name": name}

201表示已创建的状态码。

但我们没有必要记住所有代码的含义。

可以使用fastapi.status中的快捷变量。

from fastapi import FastAPI, statusapp = FastAPI()@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):return {"name": name}

更改默认状态码

高级用户指南中,将介绍如何返回与在此声明的默认状态码不同的状态码。

表单数据

接收的不是JSON,而是表单字段时,要使用Form

要使用表单,需预先安装python-multipart
例如,pip install python-multipart

步骤如下:

  • 导入Form
  • 定义Form参数:创建表单参数的方式与BodyQuery一样
from fastapi import FastAPI, Formapp = FastAPI()@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):return {"username": username}

例如,OAuth2规范的”密码流“模式规定要通过表单字段发送usernamepassword

该规范要求字段必须命名为usernamepassword,并通过表单字段发送,不能用JSON。

使用Form可以声明与Body(及QueryPathCookie)相同的元数据和验证。

Form是直接继承自Body的类

关于表单字段

在这里插入图片描述

请求文件

File用于定义客户端的上传文件。

因为上传文件以【表单数据】形式发送。所以接收上传文件,要预先安装python-multipart

步骤如下:

  • 导入File:从fastapi导入FileUploadFile
  • 定义File参数:创建文件File参数的方式与BodyForm一样
from fastapi import FastAPI, File, UploadFileapp = 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):return {"filename": file.filename}

File是直接继承自Form的类

文件作为【表单数据】上传。

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

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

不过,很多情况下,UploadFile更好用。

UploadFile的文件参数

定义文件参数时使用UploadFile

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

UploadFilebytes相比有很多优势:

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

UploadFile的属性如下:

  • filename:上传文件名字符串(str),例如,myimage.jpg
  • content_type:内容类型(MIME类型/媒体类型)字符串(str),例如,image/jpeg
  • file:SpooledTemporaryFile(file-like对象)。其实就是Python文件,可直接传递给其他预期file-like对象的函数或支持库。

UploadFile支持以下async方法,(使用内部SpooledTemporaryFile)可调用相应的文件方法。

  • write(data):把datastrbytes)写入文件;
  • 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()

在这里插入图片描述

什么是【表单数据】

与JSON不同,HTML表单(<form></form>)向服务器发送数据通常使用【特殊】的编码。

FastAPI要确保从正确的位置读取数据,而不是读取JSON。
在这里插入图片描述

可选文件上传

您可以通过使用标准类型注解并将None作为默认值的方式将一个文件参数设为可选:

from fastapi import FastAPI, File, UploadFileapp = FastAPI()@app.post("/files/")
async def create_file(file: bytes | None = File(default=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}

带有额外元数据的UploadFIle

您也可以将File()UploadFile一起使用,例如,设置额外的元数据:

from fastapi import FastAPI, File, UploadFileapp = FastAPI()@app.post("/files/")
async def create_file(file: bytes = File(description="A file read as bytes")):return {"file_size": len(file)}@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(description="A file read as UploadFile"),
):return {"filename": file.filename}

多文件上传

FastAPI支持同时上传多个文件。

可用同一个【表单字段】发送含多个文件的【表单数据】。

上传多个文件时,要声明含bytesUploadFile的列表(List):

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponseapp = FastAPI()@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.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)

接收的也是含bytesUploadFile的列表(list)。

带有额外元数据的多文件上传

和之前的方式一样,您可以为File()设置额外参数,即使是UploadFile

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponseapp = FastAPI()@app.post("/files/")
async def create_files(files: list[bytes] = File(description="Multiple files as bytes"),
):return {"file_sizes": [len(file) for file in files]}@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile] = File(description="Multiple files as 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)

请求表单与文件

FastAPI支持同时使用FileForm定义文件和表单字段。

  • 导入FileForm
  • 定义FileForm参数
from fastapi import FastAPI, File, Form, UploadFileapp = FastAPI()@app.post("/files/")
async def create_file(file: bytes = File(), fileb: UploadFile = File(), token: str = Form()
):return {"file_size": len(file),"token": token,"fileb_content_type": fileb.content_type,}

文件和表单字段作为表单数据上传与接收。

声明文件可以使用bytesUploadFile
在这里插入图片描述

处理错误

某些情况下,需要向客户端返回错误提示。

这里所谓的客户端包括前端浏览器、其他应用程序、物联网设备等。

需要向客户端返回错误提示的场景主要如下:

  • 客户端没有执行操作的权限
  • 客户端没有访问资源的权限
  • 客户端要访问的项目不存在
  • 等等…

遇到这些情况时,通常要返回4XX(400至499)HTTP状态码

使用HTTPException

向客户端返回HTTP错误响应,可以使用HTTPException

步骤如下:

  • 导入HTTPException
  • 触发HTTPExceptionHTTPException是额外包含了和API有关数据的常规Python异常。因为是Python异常,所以不能return,只能raise。如在调用路径操作函数里的工具函数时,触发了HTTPException,FastAPI就不再继续执行路径操作函数中的后续代码,而是立即终止请求,并把HTTPException的HTTP错误发送给客户端。
  • 本例中,客户端用ID请求的item不存在时,触发状态码为404的异常:
from fastapi import FastAPI, HTTPExceptionapp = 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://example.com/items/fooitem_id 为 「foo」)时,客户端会接收到HTTP状态码-200及如下JSON响应结果:

{"item": "The Foo Wrestlers"
}

但如果客户端请求 http://example.com/items/baritem_idbar」 不存在时),则会接收到 HTTP 状态码 - 404(「未找到」错误)及如下 JSON 响应结果:

{"detail": "Item not found"
}

添加自定义响应头

有些场景下要为HTTP错误添加自定义响应头。例如,出于某些方面的安全需要。

一般情况下可能不会需要在代码中直接使用响应头。

但对于某些高级应用场景,还是需要添加自定义响应头:

from fastapi import FastAPI, HTTPExceptionapp = FastAPI()items = {"foo": "The Foo Wrestlers"}@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]}

安装自定义异常处理器

添加自定义处理器,要使用 Starlette 的异常工具。

假设要触发的自定义异常叫做UnicornException

且需要FastAPI实现全局处理该异常。

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

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponseclass UnicornException(Exception):def __init__(self, name: str):self.name = nameapp = FastAPI()@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}

请求/unicorns/yolo时,路径操作会触发UnicornException

但该异常将会被unicorn_exception_handler处理。

接收到的错误信息清晰明了,HTTP状态码为418,JSON内容如下:

{"message": "Oops! yolo did something. There goes a rainbow..."}

覆盖默认异常处理器

FastAPI自带了一些默认异常处理器。

触发HTTPException或请求无效数据时,这些处理器返回默认的JSON响应结果。

不过,也可以使用自定义处理器覆盖默认异常处理器。

覆盖请求验证异常

请求中包含无效数据,FastAPI内部会触发RequestValidationError

该异常也内置了默认异常处理器。

覆盖默认异常处理器时需要导入RequestValidationError,并用@app.excption_handler(RequestValidationError)装饰异常处理器。

这样,异常处理器就可以接收Request与异常。

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

访问/items/foo,可以看到默认的JSON错误信息:

{"detail": [{"loc": ["path","item_id"],"msg": "value is not a valid integer","type": "type_error.integer"}]
}

被替换为了以下文本格式的错误信息:

1 validation error
path -> item_idvalue is not a valid integer (type=type_error.integer)

路径操作配置

路径操作装饰器支持多种配置参数。

status_code状态码

status_code 用于定义路径操作响应中的 HTTP 状态码。

可以直接传递 int 代码, 比如 404

如果记不住数字码的涵义,也可以用 status 的快捷常量:

from typing import Set, Unionfrom fastapi import FastAPI, status
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):return item

tags参数

tags参数的值是由str组成的list(一般只有一个str),tags用于为路径操作添加标签:

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(item: Item):return item@app.get("/items/", tags=["items"])
async def read_items():return [{"name": "Foo", "price": 42}]@app.get("/users/", tags=["users"])
async def read_users():return [{"username": "johndoe"}]

OpenAPI 概图会自动添加标签,供 API 文档接口使用:
在这里插入图片描述

summarydescription参数

路径装饰器还支持summarydescription这两个参数:

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/",response_model=Item,summary="Create an item",description="Create an item with all the information, name, description, price, tax and a set of unique tags",
)
async def create_item(item: Item):return item

文档字符串(docstring

描述内容比较长且占用多行时,可以在函数的docstring中声明路径操作的描述,FastAPI支持从文档字符串中读取描述内容。

文档字符串支持Markdown,能正确解析和显示Markdown的内容,但要注意文档字符串的缩进。

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(item: Item):"""Create an item with all the information:- **name**: each item must have a name- **description**: a long description- **price**: required- **tax**: if the item doesn't have tax, you can omit this- **tags**: a set of unique tag strings for this item"""return item

下图为 Markdown 文本在 API 文档中的显示效果:
在这里插入图片描述

响应描述

response_description参数用于定义响应的描述说明:

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/",response_model=Item,summary="Create an item",response_description="The created item",
)
async def create_item(item: Item):"""Create an item with all the information:- **name**: each item must have a name- **description**: a long description- **price**: required- **tax**: if the item doesn't have tax, you can omit this- **tags**: a set of unique tag strings for this item"""return item

弃用路径操作

deprecated参数可以把路径操作标记为弃用,无需直接删除:

from fastapi import FastAPIapp = FastAPI()@app.get("/items/", tags=["items"])
async def read_items():return [{"name": "Foo", "price": 42}]@app.get("/users/", tags=["users"])
async def read_users():return [{"username": "johndoe"}]@app.get("/elements/", tags=["items"], deprecated=True)
async def read_elements():return [{"item_id": "Foo"}]

JSON兼容编码器

在某些情况下,您可能需要将数据类型(如Pydantic模型)转换为与JSON兼容的数据类型(如dictlist)。

比如,如果您需要将其存储在数据库中。

对于这种要求,FastAPI提供了jsonable_encoder()函数。

使用jsonable_encoder

让我们假设你有一个数据库名为fake_db,它只能接收与JSON兼容的数据。

例如,它不接收datatime这类的对象,因为这些对象与JSON不兼容。

因此,datatime对象必须将转换为包含ISO格式化的str类型对象。

同样,这个数据库也不会接收Pydantic模型(带有属性的对象),而只接受dict

对此你可以使用jsonable_encoder

它接收一个对象,比如Pydantic模型,并会返回一个JSON兼容的版本:

from datetime import datetimefrom fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModelfake_db = {}class Item(BaseModel):title: strtimestamp: datetimedescription: str | None = Noneapp = FastAPI()@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

在这个例子中,它将Pydantic模型转换为dict,并将datetime转换为str

调用它的结果后就可以使用Python标准编码中的json.dumps()。

这个操作不会返回一个包含JSON格式(作为字符串)数据的庞大的str。它将返回一个Python标准数据结构(例如dict),其值和子值都与JSON兼容。

请求体-更新数据

PUT更新数据

更新数据请用HTTP PUT操作。

把输入数据转换为以JSON格式存储的数据(比如,使用NoSQL数据库时),可以使用jsonable_encoder。例如,把datetime转换为str

from typing import List, Unionfrom fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: Union[str, None] = Nonedescription: Union[str, None] = Noneprice: Union[float, None] = Nonetax: float = 10.5tags: 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)
async def read_item(item_id: str):return items[item_id]@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):update_item_encoded = jsonable_encoder(item)items[item_id] = update_item_encodedreturn update_item_encoded

PUT用于接收替换现有数据的数据。

PUT把数据项bar更新为以下内容时:

{"name": "Barz","price": 3,"description": None,
}

因为上述数据未包含已存储的属性 "tax": 20.2,新的输入模型会把 "tax": 10.5 作为默认值。

因此,本次操作把 tax 的值「更新」为 10.5。

PATCH进行部分更新(用PUT也无妨)

  • 使用Pydantic的exclude_unset参数:更新部分数据时,可以在Pydantic模型的.dict()中使用exclude_unset参数。比如item.dict(exclude_unset=True)。这段代码生成的dict只包含创建item模型时显式设置的数据,而不包括默认值。然后再用它生成一个只含已设置(在请求中发送)数据,且省略了默认值的dict
  • 使用Pydantic的update参数:接下来,用.copy()为已有模型创建调用update参数的副本,该参数为包含更新数据的dict。例如,stored_item_model.copy(update=update_data)
from typing import List, Unionfrom fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: Union[str, None] = Nonedescription: Union[str, None] = Noneprice: Union[float, None] = Nonetax: float = 10.5tags: 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)
async def read_item(item_id: str):return items[item_id]@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

更新部分数据小结
简而言之,更新部分数据应:

  • 使用PATCH而不是PUT(可选,也可以用PUT);
  • 提取存储的数据;
  • 把数据放入Pydantic模型;
  • 生成不含输入模型默认值的dict(使用exclude_unset参数);只更新用户设置过的值,不用模型中的默认值覆盖已存储过的值。
  • 为已存储的模型创建副本,用接收的数据更新其属性(使用update参数)。
  • 把模型副本转换为可存入数据库的形式(比如,使用jsonable_encoder)。把这种方式与Pydantic模型的.dict()方法类似,但能确保把值转换为适配JSON的数据类型,例如,把datetime转换为str
  • 把数据保存至数据库;
  • 返回更新后的模型。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/pingmian/37874.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

HIVE每日一题

select * from sku_info order by sku_id ; 为什么结果没有顺序排序。什么原因导致的&#xff1f;

小红书多账号管理平台哪个好用?可以快速监测多个小红书账号的数据吗?

随着品牌营销战线的不断扩展&#xff0c;小红书已经成为企业和个人品牌竞相展示的舞台。但是&#xff0c;随之而来的多账号管理问题也让众多运营者头疼不已。一个优秀的多账号管理平台&#xff0c;能让你事半功倍&#xff0c;轻松监控和分析账号数据。 如今&#xff0c;市面上出…

SiteSucker Pro for Mac:一键下载整站,轻松备份与离线浏览!

SiteSucker Pro for Mac是一款专为苹果电脑用户设计的网站下载与备份工具&#x1f578;️。它以其强大的整站下载能力和用户友好的界面&#xff0c;成为了众多Mac用户备份网站、离线浏览的得力助手&#x1f4bb;。 这款软件允许用户一键下载整个网站&#xff0c;包括所有的网页…

鸿蒙开发Ability Kit(程序框架服务):【选择申请权限的方式】

选择申请权限的方式 应用在访问数据或者执行操作时&#xff0c;需要评估该行为是否需要应用具备相关的权限。如果确认需要目标权限&#xff0c;则需要在应用安装包中申请目标权限。 每一个权限的权限等级、授权方式不同&#xff0c;申请权限的方式也不同&#xff0c;开发者在…

java spring 拦截器和过滤器+过滤器处理数据

java spring 拦截器和过滤器过滤器处理数据 介绍Spring拦截器&#xff08;Interceptor&#xff09;导入依赖完整代码 Java过滤器&#xff08;Filter&#xff09;完整代码分为 2 个文件&#xff0c;请看下面BodyReaderHttpServletRequestWrapper.javaMyFilter.java Spring过滤器…

DataWhale-吃瓜教程学习笔记 (五)

学习视频&#xff1a;第4章-决策树_哔哩哔哩_bilibili 西瓜书对应章节&#xff1a; 第四章 4.1&#xff1b;4.2 文章目录 决策树算法原理- 逻辑角度- 几何角度 ID3 决策树- 自信息- 信息熵 &#xff08;自信息的期望&#xff09;- 条件熵 &#xff08; Y 的信息熵关于概率分布 …

python selenium 打开网页

selenium工具类 - 文件名 seleniumkit.py 代码如下 # -*- coding:utf-8 _*-from selenium import webdriverimport os import timefrom selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from seleniu…

Ascend基于自定义算子工程的算子开发

环境准备 见https://gitee.com/zaj1414904389/ascend-tutorial.git 工程创建 CANN软件包中提供了工程创建工具msopgen&#xff0c;开发者可以输入算子原型定义文件生成Ascend C算子开发工程 [{"op": "AddCustom","input_desc": [{"name…

CVPR2024 轨迹预测系列(一)

CVPR2024 轨迹预测系列&#xff08;一&#xff09; 文章目录 CVPR2024 轨迹预测系列&#xff08;一&#xff09;一、Adapting to Length Shift: FlexiLength Network for Trajectory Prediction.1、Abstract2、FlexiLength Network3、Datasets4、Experiments5、Implementation …

分类预测 | PSO-PNN基于粒子群算法优化概率神经网络的数据分类预测(Matlab)

分类预测 | ZOA-PCNN-AT-SVM斑马优化并行卷积-支持向量机融合注意力机制的故障识别 目录 分类预测 | ZOA-PCNN-AT-SVM斑马优化并行卷积-支持向量机融合注意力机制的故障识别分类效果基本描述程序设计参考资料 分类效果 基本描述 1.PSO-PNN基于粒子群算法优化概率神经网络的数据…

长鑫存储母公司斥资24亿美元发展国产HBM

国产DRAM厂商长鑫存储母公司睿力集成计划投资24亿美元在上海建一座高端封装工厂。据报道&#xff0c;该工厂将专注于高带宽存储器&#xff08;HBM&#xff09;芯片的封装&#xff0c;预计到2026年中开始投入生产。长鑫存储将利用来自多方投资者的资金进行建设&#xff0c;其中包…

鼠尾草(洋苏草)

鼠尾草&#xff08;Salvia japonica Thunb.&#xff09;&#xff0c;又名洋苏草、普通鼠尾草、庭院鼠尾草&#xff0c;属于唇形科鼠尾草属多年生草本植物。鼠尾草以其独特的蓝紫色花序和长而细密的叶片为特点&#xff0c;常用于花坛、庭院和药用植物栽培。 鼠尾草的名字源自于…

江科大笔记—读写内部闪存FLASH读取芯片ID

读写内部闪存FLASH 右下角是OLED&#xff0c;然后左上角在PB1和PB11两个引脚&#xff0c;插上两个按键用于控制。下一个代码读取芯片ID&#xff0c;这个也是接上一个OLED&#xff0c;能显示测试数据就可以了。 STM32-STLINK Utility 本节的代码调试&#xff0c;使用辅助软件…

Langchain-实战篇-搭建本地问答机器人-01

项目背景 在这个快节奏的职场世界里&#xff0c;我们每个人都可能在某个时刻&#xff0c;化身为一头辛勤的牛或一匹奔腾的马&#xff0c;面对入职签合同时的迷茫&#xff0c;或是离职时的纠纷&#xff0c;心中难免会涌现出各种疑问。比如&#xff1a; "这份合同里的条款…

Python在Excel中设置数字格式和获取应用数字格式后的值

目录 安装Python Excel库 Python在Excel中设置数字格式 Python获取Excel中应用数字格式的单元格的显示值 总结 Excel 数字格式是用于控制单元格中数字显示方式的一组规则或代码。通过设置不同的数字格式&#xff0c;可以定义数字的显示方式&#xff0c;如小数位数、货币符号…

Java对应C++ STL的用法

sort&#xff1a; 1&#xff1a;java.util.Arrays中的静态方法Arrays.sort()方法&#xff0c;针对基本数据类型和引用对象类型的数组元素排序 2&#xff1a;java.util.Collections中的静态方法的Collections.sort()方法&#xff0c;针对集合框架中的动态数组&#xff0c;链表&…

iOS17系统适配

iOS17 新功能 文章目录 iOS17 新功能iOS17支持哪几款机型Xcode15新特性iOS17-开发适配指南 横屏待机 在iOS 17中&#xff0c;还带来了横屏待机功能&#xff0c;苹果将这个新功能命名为“Standby”模式&#xff0c;为 iPhone 带来了全新的玩法。iPhone启用之后&#xff0c;默认情…

1-Pandas是什么

Pandas是什么 Pandas 是一个开源的第三方 Python 库&#xff0c;从 Numpy 和 Matplotlib 的基础上构建而来&#xff0c;享有数据分析“三剑客之一”的盛名&#xff08;NumPy、Matplotlib、Pandas&#xff09;。Pandas 已经成为 Python 数据分析的必备高级工具&#xff0c;它的…

PyTorch Tensor进阶操作指南(二):深度学习中的关键技巧

本文主要讲tensor的裁剪、索引、降维和增维 Tensor与numpy互转、Tensor运算等&#xff0c;请看这篇文章 目录 9.1、首先看torch.squeeze()函数&#xff1a; 示例9.1&#xff1a;&#xff08;基本的使用&#xff09; 小技巧1&#xff1a;如何看维数 示例9.2&#xff1a;&a…

安全架构概述_1.信息安全面临的威胁

在当今以计算机、网络和软件为载体的数字化服务几乎成为人类赖以生存的手段。与之而来的计算机犯罪呈现指数上升趋势&#xff0c;因此&#xff0c;信息的可用性、完整性、机密性、可控性和不可抵赖性等安全保障显得尤为重要&#xff0c;而满足这些诉求&#xff0c;离不开好的安…