全栈开发

FastAPI官方文档学习02

Posted by

1、Cookie

@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
    return {"ads_id": ads_id}
  • ads_id参数 属于Cookie参数
    • 类似于路径参数 Path（从路径中获取数据） 查询参数 Query（从查询参数中获取数据）
    • Cookie()从Cookie中获取对应参数名的cookie 键的值

2、Header

@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):
    return {"User-Agent": user_agent}
  • user_agent参数 属于Header参数，从Header中获取对应参数名的Header的属性的值
  • 默认情况下，Header 把参数名中的字符由下划线（_）改为连字符（-）来提取并存档请求头
  • 如需禁用下划线自动转换为连字符，可以把 Header 的 convert_underscores 参数设置为 False

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

重复的请求头

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

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

3、Cookie参数模型

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
  • FastAPI 将从请求中接收到的 cookie 中提取出每个字段的数据，并提供您定义的 Pydantic 模型。

4、Header参数模型

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

5、响应模型

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: list[str] = []



@app.get("/items/", response_model=list[Item])
async def read_items() -> Any:
    return [
        {"name": "Portal Gun", "price": 42.0},
        {"name": "Plumbus", "price": 32.0},
    ]

# 请求体类型 和 响应体类型都是Item类型
@app.post("/items/", response_model=Item)
async def create_item(item: Item) -> Any:
    return item

FastAPI 将使用此 response_model 来：

  • 将输出数据转换为其声明的类型。
  • 校验数据。

重要的是：将输出数据限制在该模型定义内

添加输出模型

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

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

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]

6、更多模型

class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None


class UserIn(UserBase):
    password: str


class UserOut(UserBase):
    pass

class UserInDB(UserBase):
    hashed_password: str


def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password

def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.model_dump(), 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)
    print(user_saved)
    return user_saved

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

class BaseItem(BaseModel):
    description: str
    type: str


class CarItem(BaseItem):
    type: str = "car"


class PlaneItem(BaseItem):
    type: str = "plane"
    size: int


items = {
    "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]

模型列表

class Item(BaseModel):
    name: str
    description: str


items = [
    {"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 模型。

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

7、响应状态码

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

app = FastAPI()
fastapi_cdn_host.patch_docs(app)


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


@app.post("/users/", status_code=status.HTTP_201_CREATED)
async def create_user(name: str):
    return {"name": name}

8、表单数据

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

创建表单（Form）参数的方式与 Body 和 Query 一样：

from fastapi import FastAPI, Form

@app.post("/login/")
async def login(username: Annotated[str, Form()], password: Annotated[str, Form()]):
    return {"username": username}
  • 使用 Form 可以声明与 Body （及 Query、Path、Cookie）相同的元数据和验证。
  • Form 是直接继承自 Body 的类。
  • 声明表单体要显式使用 Form ，否则，FastAPI 会把该参数当作查询参数或请求体（JSON）参数。

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

但包含文件的表单编码为 multipart/form-data。

9、表单模型

前置依赖

pip install python-multipart
class FormData(BaseModel):
    username: str
    password: str


@app.post("/login")
async def login(data: Annotated[FormData, Form()]):
    return data

10、请求文件

from typing import Annotated
from fastapi import FastAPI, File, UploadFile
import fastapi_cdn_host

app = FastAPI()
fastapi_cdn_host.patch_docs(app)

@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 的类。
  • 如果把路径操作函数参数的类型声明为 bytes，FastAPI 将以 bytes 形式读取和接收文件内容。

UploadFile 与 bytes 相比有更多优势：

  • 使用 spooled 文件：
    • 存储在内存的文件超出最大上限时，FastAPI 会把文件存入磁盘；
  • 这种方式更适于处理图像、视频、二进制文件等大型文件，好处是不会占用所有内存；
  • 可获取上传文件的元数据；
  • 自带 file-like async 接口；
  • 暴露的 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)：把 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: Annotated[bytes | None, File()] = None):
    return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile | None = None):
    return {"filename": file.filename}

多文件上传

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

11、请求表单与文件

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

from typing import Annotated
from fastapi import FastAPI, File, Form, UploadFile
import fastapi_cdn_host

app = FastAPI()
fastapi_cdn_host.patch_docs(app)


@app.post("/files/")
async def create_file(
    file: Annotated[bytes, File()],
    fileb: Annotated[UploadFile, File()],
    token: Annotated[str, Form()]
):
    return {
        "file_size": len(file),
        "token": token,
        "file_content_type": fileb.content_type
    }

声明文件可以使用 bytes 或 UploadFile 。

下一篇
业务域划分五步实操法
上一篇 团队数据治理规范