Fastapi + vue3 自动化测试平台（4）-- fastapi分页查询封装

在构建自动化测试平台时，分页查询是一个非常常见的功能。本文将以 FastAPI 为例，介绍如何封装一个通用的分页查询功能，使其更便于复用。

需求背景

在测试平台中，我们可能需要对用户操作记录、测试用例、任务日志等数据进行分页展示。通过封装分页查询逻辑，可以避免重复编写分页代码，提高开发效率，同时增强代码的可维护性。

实现思路

分页查询的核心步骤包括：

接收前端传递的分页参数（如 currentPage 和 pageSize）。
根据查询条件筛选数据。
计算分页偏移量。
查询数据并统计总数。
返回结构化的分页结果。

优点分析：

支持动态查询条件：通过 search_dict 构建动态查询条件，使得功能更灵活。
分页和排序：实现了分页和排序功能，适用于大部分场景。
关联数据处理：通过 user_id 关联查询用户名，增强了数据的可读性。
数据清洗：通过 remove_key 和 time_to_str 对数据进行了处理，提升了返回结果的质量。

工具函数

# 首先，我们封装一个通用的分页查询方法：
import json
from re import search
from datetime import datetime
from common.request_to_json import body_to_json
from common.time_str import time_to_str
from views.user.user_model import User_info# 分页查询
async def db_page_all(db, request, key, order):body = await body_to_json(request)search_dict = {k: v for k, v in body["search"].items() if v != ''}db_select = db.filter(**search_dict)data = await db_select.offset((body["currentPage"] - 1) * body["pageSize"]).limit(body["pageSize"]).order_by(order).values()res = await remove_key(data, key)result = await time_to_str(res)count = await db_select.count()return {"content": result,"currentPage": body["currentPage"],"pageSize": body["pageSize"],"total": count}# 处理请求体
async def body_to_json(request):data = await request.body()return json.loads(data)# 剔除不想展示得key
async def remove_key(data, key):result = []for i in data:var = {k: v for k, v in i.items() if k not in key}result.append(var)return result# 处理时区的时间格式
async def time_to_str(data):if isinstance(data, dict):if "user_id" in data.keys():user = await User_info.get(id=data["user_id"])data["username"] = user.usernameif "create_time" in data.keys():data["create_time"] = datetime.strftime(data["create_time"], "%Y-%m-%d %H:%M:%S")if "update_time" in data.keys():data["update_time"] = datetime.strftime(data["update_time"], "%Y-%m-%d %H:%M:%S")if "end_time" in data.keys():data["end_time"] = datetime.strftime(data["end_time"], "%Y-%m-%d %H:%M:%S")if "start_time" in data.keys():data["start_time"] = datetime.strftime(data["start_time"], "%Y-%m-%d %H:%M:%S")if "token_time" in data.keys():data["token_time"] = datetime.strftime(data["token_time"], "%Y-%m-%d %H:%M:%S")elif isinstance(data, list):for i in data:if "user_id" in i.keys():user = await User_info.get(id=i["user_id"])i["username"] = user.usernameif "create_time" in i.keys():i["create_time"] = datetime.strftime(i["create_time"], "%Y-%m-%d %H:%M:%S")if "update_time" in i.keys():i["update_time"] = datetime.strftime(i["update_time"], "%Y-%m-%d %H:%M:%S")if "token_time" in i.keys():i["token_time"] = datetime.strftime(i["token_time"], "%Y-%m-%d %H:%M:%S")if "end_time" in i.keys():i["end_time"] = datetime.strftime(i["end_time"], "%Y-%m-%d %H:%M:%S")if "start_time" in i.keys():i["start_time"] = datetime.strftime(i["start_time"], "%Y-%m-%d %H:%M:%S")return data

# 调用函数
from views.user.user_model import User_info
from fastapi import Request@router.post("/user_list")
async def user_list(request: Request):try:data = await db_page_all(User_info, request, key=["token", "token_time", "password", "account"], order="-id")return await res_success(data)except Exception as e:return await catch_Exception(e)

结果展示

{"content": [{"id": 1,"username": "test_user","action": "login","timestamp": "2025-01-22 12:34:56"},{"id": 2,"username": "another_user","action": "logout","timestamp": "2025-01-21 15:45:23"}],"currentPage": 1,"pageSize": 10,"total": 2}