当前位置： 首页 > news >正文

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

news 2025/8/4 7:31:59

响应模型

添加响应模型

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="john.doe@example.com")# **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 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 & 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 commons@app.get("/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

http://www.lryc.cn/news/400377.html

相关文章：

  • 案例 | 人大金仓助力山西政务服务核心业务系统实现全栈国产化升级改造
  • 如何用python写接口
  • 轻量级可扩展易航网址引导系统源码V2.45
  • 解决ESLint和Prettier冲突的问题
  • C判断一个点在三角形上
  • 物业系统自主研发接口测试框架
  • 手机和电脑通过TCP传输
  • Git 在commit后，撤销commit
  • 多模态大模型 - MM1
  • FPGA设计之跨时钟域（CDC）设计篇（2）----如何科学地设计复位信号？
  • GPS北斗标准时钟同步服务器结构是什么？安徽京准
  • 9.5 栅格图层符号化多波段彩色渲染
  • 力扣第九题
  • 鞭炮插画：成都亚恒丰创教育科技有限公司
  • python 循环
  • 映美精黑白相机IFrameQueueBuffer转halcon的HObject
  • Linux的load（负载）
  • 杜比全景声——空间音频技术
  • C 语言指针进阶
  • SpringBootWeb 篇-入门了解 Swagger 的具体使用
  • Python面试题：如何在 Python 中处理大数据集？
  • C++:入门基础
  • 微信小游戏 彩色试管 倒水游戏 逻辑 （二）
  • 【链表】算法题（一） ---- 力扣 / 牛客
  • Linux系统之部署盖楼小游戏
  • “金山-讯飞”杯2024年武汉理工大学程序设计竞赛 A. Mobiusp败走***（思维题-点双连通分量、连通性）
  • 【机器翻译】基于术语词典干预的机器翻译挑战赛
  • 推荐系统：从协同过滤到深度学习
  • 记录些Spring+题集（1）
  • SpringBoot 解决 getSession().getAttribute() 在负载均衡环境下无法获取session的问题