FastAPI -- 第二弹(响应模型、状态码、路由APIRouter、后台任务BackgroundTasks)

响应模型

添加响应模型

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# 通过 response_model 设置返回模型
@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:return user

模型的其他操作

继承 & 解包

from fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp = FastAPI()# 定义一个 Base
# 包含公共属性，比如 id create_time update_time 等
class UserBase(BaseModel):username: stremail: EmailStrfull_name: str | None = None# 继承 Base
class UserIn(UserBase):password: str# 继承 Base
class UserOut(UserBase):pass# 继承 Base
class 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 = UserIn(username="john", password="secret", email="[e-mail]")# **user_in.dict() ==> #		username = user_dict["username"],#	    password = user_dict["password"],#	    email = user_dict["email"],#	    full_name = user_dict["full_name"], user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)print("User saved! ..not really")return [e-mail]("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):user_saved = fake_save_user(user_in)return user_saved

组合模型 （Union & dict）

union

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,},"item3": [{"description": "item3_1 Music is my aeroplane, it's my aeroplane","type": "plane","size": 5,},{"description": "item3_2 Music is my aeroplane, it's my aeroplane","type": "plane","size": 5,},]
}# Union[PlaneItem, CarItem, list[PlaneItem]]，。。。。]
# 允许返回 Union 的多种类型中的 一种
@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem,list[PlaneItem]])
async def read_item(item_id: str):return items[item_id]

dict

response_model=dict ,
response_model=dict[str, list] ,
response_model=list …

dict 有点类似于 any， 但又比 any 更具体一些
比如 any 可以是 list 也可以是 dict， 但 dict 只能是 dict 不能是 list
比如 dict 可以是 dict[str, list]， dict[int, list] 等，但 dict[str, list] 只能是 dict[str, list]

from fastapi import FastAPIapp = FastAPI()# 测试 Any
@app.get("/test_any/")
async def read_keyword_weights():# return {"foo": 2.3, "bar": 3.4}return [1, 2, 3]# 测试 dict
@app.get("/test_dict/", response_model=dict)
async def read_keyword_weights():return {"foo": 2.3, "bar": [1, 2, 3]}# 测试 dict[str, float]
@app.get("/test_dict_str_float/", response_model=dict[str, float])
async def read_keyword_weights():return {"foo": 2.3, "bar": 3.4}# return {"foo": 2.3, "bar": [1, 2, 3]}

响应状态码

from enum import IntEnum
from http import HTTPStatusfrom fastapi import FastAPI, statusapp = FastAPI()# 通过 http模块的 HTTPStatus 类，
# 我们可以很方便的使用一些常见的状态码，如下
@app.get("/items/http_status", status_code=HTTPStatus.OK)
def test_status_code():return "hallow world HTTPStatus"# 通过 fastapi 的 status 模块，
# 我们可以很方便的使用一些常见的状态码，如下
@app.get("/items/status", status_code=status.HTTP_201_CREATED)
def test_status_code():return "hallow world status"# 也可以直接使用数字
@app.get("/items/num", status_code=201)
def test_status_code():return "hallow world num"

顺便记一下，可以仿照 HTTPStatus 的写法，自定义状态码。
注意：自定义状态码是放到 响应体 里的，放到 status_code 后面有可能会报错哦

# 自定义的状态码
class MyHTTPStatus(IntEnum):    def __new__(cls, value, phrase, description=''):obj = int.__new__(cls, value)obj._value_ = valueobj.phrase = phraseobj.description = descriptionreturn objCode200000 = 200000, 'OK', 'Request fulfilled, document follows'# client errorCode400000 = (400000, 'Bad Request','Bad request syntax or unsupported method')# 响应体示例
# {
#     "code": 200000,
#     "message": "OK",
#     "data": {
#         "name": "张三"
#     }
# }

From表单

from fastapi import FastAPI, Formapp = FastAPI()# 和 Boday Query 类似，Form 也是从 fastapi 导入
# username: str = Form()
@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):return {"username": username}

上传文件 (File, UploadFile)

from fastapi import FastAPI, File, UploadFileapp = FastAPI()@app.post("/files/")
async def create_file(file: bytes = File()):return {"file_size": len(file)}# 和前面的可选参数类似，如果 file 是可选的可以使用如下的方式声明
# file: UploadFile | None = None
# 多文件上传
# files: list[UploadFile]
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):return {"filename": file.filename}

UploadFile 与 bytes 相比有更多优势：

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

UploadFile

UploadFile 的属性

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

UploadFile 支持以下 async 方法

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

依赖项

声明依赖项

• 提高代码的复用性

from typing import Unionfrom fastapi import Depends, FastAPIapp = FastAPI()# 为查询接口设置一些公共参数
async def common_parameters(q: Union[str, None] = None, skip: int = 0, limit: int = 100
):return {"q": q, "skip": skip, "limit": limit}# 通过 Depends 声明依赖项
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):return [e-mail]("/users/")
async def read_users(commons: dict = Depends(common_parameters)):return commons

APIRouter

相当于 Flask 的 Blueprints

目录结构

.
├── app                  # 「app」是一个 Python 包
│   ├── __init__.py      # 这个文件使「app」成为一个 Python 包
│   ├── main.py          # 「main」模块，例如 import app.main
│   ├── dependencies.py  # 「dependencies」模块，例如 import app.dependencies
│   └── routers          # 「routers」是一个「Python 子包」
│   │   ├── __init__.py  # 使「routers」成为一个「Python 子包」
│   │   ├── items.py     # 「items」子模块，例如 import app.routers.items
│   │   └── users.py     # 「users」子模块，例如 import app.routers.users
│   └── internal         # 「internal」是一个「Python 子包」
│       ├── __init__.py  # 使「internal」成为一个「Python 子包」
│       └── admin.py     # 「admin」子模块，例如 import app.internal.admin

app/dependencies.py

from fastapi import Header, HTTPExceptionasync def get_token_header(x_token: str = Header()):if x_token != "fake-super-secret-token":raise HTTPException(status_code=400, detail="X-Token header invalid")async def get_query_token(token: str):if token != "jessica":raise HTTPException(status_code=400, detail="No Jessica token provided")

app/routers/items.py

from fastapi import APIRouter, Depends, HTTPException# 可以使用相对路径进行导入操作
# 但个人不推荐这样操作
from ..dependencies import get_token_header# 定义 APIRouter（定义模块 / 定义蓝图）
router = APIRouter(prefix="/items",tags=["items"],dependencies=[Depends(get_token_header)],responses={404: {"description": "Not found"}},
)fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}# prefix + path
# api ==》 /items/
@router.get("/")
async def read_items():return fake_items_db# prefix + path
# api ==》 /items/{item_id}
@router.get("/{item_id}")
async def read_item(item_id: str):if item_id not in fake_items_db:raise HTTPException(status_code=404, detail="Item not found")return {"name": fake_items_db[item_id]["name"], "item_id": item_id}@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"}

app/main.py

from fastapi import Depends, FastAPIfrom .dependencies import get_query_token, get_token_header
from .internal import admin# 导入模块路由
from .routers import items, usersapp = 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

from fastapi import BackgroundTasks, FastAPIapp = 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: BackgroundTasks# 然后通过 background_tasks.add_task ，添加为后台任务# 并没有看到 background_tasks 创建，# 应该是 FastAPI 会创建一个 BackgroundTasks 类型的对象并作为该参数传入background_tasks.add_task(write_notification, email, message="some notification")return {"message": "Notification sent in the background"}

---

到此结  DragonFangQy 2024.07.13