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

《FastAPI零基础入门与进阶实战》第14篇:ORM之第一个案例改善-用户查询

news 2025/8/28 7:11:30

系列文章目录

《FastAPI零基础入门与进阶实战》https://blog.csdn.net/sen_shan/category_12950843.html

《FastAPI零基础入门与进阶实战》第13篇:ORM之第一个案例改善-用戶新增https://blog.csdn.net/sen_shan/article/details/150584147?spm=1001.2014.3001.5501

文章目录

目录

系列文章目录

文章目录

返回信息案例

QueryModel

CRUD

Router

参数案例

前言

        在第11篇ORM示例中，我们利用SQLAlchemy完成了数据的查询操作，但遗憾的是，这些操作并未返回状态值来指示执行结果。本章节将着重探讨如何为这些数据操作添加返回值，以实现更精准的结果反馈，提升功能的完善度。

返回信息案例

{"status_code": 200,"status": "success","message": "共计：2条数据","record_count": 2,"data": {"count": 2,"page": 2,"page_size": 2,"total_count": 12,"details": [{"is_superuser": false,"deletion_by": null,"is_staff": true,"create_date": "2025-08-20T14:14:41","id": "4a368e9b-306a-4123-8ce1-c27b07c0e1c6","effective_date": "2025-04-24","creator_by": null,"expiry_date": null,"last_updated_date": null,"last_updated_by": null,"api_id": "some-api-id","deletion_mark": false,"login_password": "some-password","user_name": "some-user-name","deletion_reason": null,"login_id": "some-login-id","email": "user@example.com","deletion_date": null,"person_id": "some-person-id"},{"is_superuser": false,"deletion_by": null,"is_staff": true,"create_date": "2025-08-20T14:48:26","id": "4a4b6673-ec46-4bf4-ab78-09fbe8616fc9","effective_date": "2025-04-24","creator_by": null,"expiry_date": null,"last_updated_date": null,"last_updated_by": null,"api_id": "some-api-id","deletion_mark": false,"login_password": "some-password","user_name": "some-user-name","deletion_reason": null,"login_id": "some-login-id","email": "user@example.com","deletion_date": null,"person_id": "some-person-id"}]}
}

QueryModel

在src\schemas目录下修改request_model.py

class QueryModel(BaseModel):page: Optional[int] = 1page_size: Optional[int] = 100data: Optional[dict] = None  # 其他过滤字段

         QueryModel  只是把上一段代码里 Query进一步用 Pydantic 模型 做了一层类型校验与文档化，方便 FastAPI 自动生成 OpenAPI/Swagger 文档，并保证运行时参数符合的 JSON规则：

CRUD

修改src\crud\sys_user.py

from typing import List, Dict, Any, Optional
def get_user(db: Session, id: str):"""根据 ID 获取用户记录。"""db_response = db.query(models.SysUser).filter(models.SysUser.id == id).first()if db_response:db_dict = [strU.model_to_dict(db_response)]  # 单条数据包装为列表else:db_dict = []  # 无数据时返回空列表response_data = {"total_count": len(db_dict),"page": 1,"page_size": 1,"details": db_dict}return retMes.Success(response_data, '共计：' + str(len(db_dict)) + '条数据', record_count=len(db_dict)).mes()# return db.query(models.SysUser).filter(models.SysUser.id == id).first()def get_users(db: Session, page: int = 0, page_size: int = 100, FilterModel: request_model.QueryModel = None,id: str = None):# FilterData: Optional[schemas.SysUser]=NoneFilterData: Optional[dict] = Nonequery = db.query(models.SysUser)if strU.is_not_empty( id):query = query.filter(models.SysUser.id == id)elif strU.is_not_empty(FilterModel):if FilterModel.page is not None :page = FilterModel.pageif  FilterModel.page_size is not None :page_size = FilterModel.page_sizeFilterData = FilterModel.data# 如果有过滤条件，则添加过滤if strU.is_not_empty(FilterData):for key, value in FilterData.items():if hasattr(models.SysUser, key) and value is not None:query = query.filter(getattr(models.SysUser, key) == value)# 应用分页并执行查询if strU.is_empty(page) or strU.is_empty(page_size):total_count = query.count()db_responses=query.all()else:if page > 1:skip =(page-1)*page_size+1else :   skip =0limit = page_sizetotal_count = query.count()db_responses = query.offset(skip).limit(limit).all()#  db_responses = db.query(models.SysUser).offset(skip).limit(limit).all()db_responses_dict = [strU.model_to_dict(db_response) for db_response in db_responses]"""if db_response:db_dict = strU.model_to_dict(db_response)else:db_dict = []"""# 将数据包装在字典中以符合 ResponseModel 要求response_data = {"count": len(db_responses_dict),"page": page,"page_size": page_size,"total_count": total_count,"details": db_responses_dict}retDatas = retMes.Success(response_data, '共计：' + str(len(db_responses_dict)) + '条数据',record_count=len(db_responses_dict)).mes()return retDatas  # db_response# return db.query(models.SysUser).offset(skip).limit(limit).all()

主要从数据库里读取用户表（models.SysUser）记录的，虽然存在区别，最终只只用get_users：
1. get_user(db, id)  • 单条查询：根据主键 id 精确查找一条用户记录。 
2. get_users(db, page=0, page_size=100, FilterModel=None, id=None)  • 批量查询 / 分页查询 / 条件过滤查询：
 – 如果传了 id，则退化为按 id 精确查询（但仍返回分页格式）。
 – 否则优先使用 FilterModel 里的 page / page_size 覆盖默认参数。
 – FilterModel.data 里的键值对会被当作等值过滤条件追加到 SQL where 子句。
 • 分页逻辑：
 – page=0 或 page_size 为 0 时，直接查全表。
 – page>1 时，skip 计算为 (page-1)*page_size+1；page=1 时 skip 为 0。
 • 返回格式：
 {  "count": 当前页实际条数,  "page": 当前页码,  "page_size": 每页条数,  "total_count": 符合条件的总条数,  "details": [ {...}, {...}, ... ]  }  • 同样再包一层 retMes.Success(...)，提示文字“共计：x 条数据”。
总结：
get_user 适用于“已知 id 查单条”的场景；get_users 不仅适用于“已知 id 查单条”的场景，更加适用于“查列表、支持分页和任意字段过滤”的场景。

Router

        修改src\Router\sys_manges.py


# 获取用户
@router.get("/users/{id}", response_model=request_model.ResponseModel)
def read_user(id: str, db: Session = Depends(get_db)):# db_user = sysUserCrud.get_user(db, id)return sysUserCrud.get_users(db, None, None, None, id)if db_user is None:raise HTTPException(status_code=404, detail="User not found")return db_user# 获取所有用户
# @router.get("/users/", response_model=list[sysUserSchema.SysUser])
@router.get("/users/", response_model=request_model.ResponseModel)
def read_users(page: int = 1, page_size: int = 100, db: Session = Depends(get_db)):# 将页码转换为skip值# skip = (page - 1) * page_sizereturn sysUserCrud.get_users(db, page, page_size)# 获取所有用户
# @router.post("/users/", response_model=list[sysUserSchema.SysUser])
@router.post("/users/query/", response_model=request_model.ResponseModel)
def read_users(queryModel: request_model.QueryModel = None, db: Session = Depends(get_db)):# 将页码转换为skip值return sysUserCrud.get_users(db, None, None, queryModel)

接口功能说明（按路径归类）
1. GET /users/{id}  • 功能：根据主键 id 精确查询一条用户记录。
 • 调用链：内部直接调用 sysUserCrud.get_users(db, None, None, None, id)。
 • 返回值：统一结构的Json
 • 注意：函数体最后的
 if db_user is None: raise HTTPException(...)
 实际不会执行，因为前面已经 return 了；因此 404 场景目前不会触发。
2. GET /users/  • 功能：查询全表并分页返回用户列表。
 • 参数：
 – page: 起始页码（默认 1）
 – page_size: 每页条数（默认 100）
 • 调用链：内部把 page 转成 skip 后调用 sysUserCrud.get_users(db, skip, page_size)。
 • 返回值：统一结构的Json。
3. POST /users/query/  • 功能：按复杂条件查询用户列表并分页/排序。
 • 请求体：request_model.QueryModel（JSON），可包含
 – page / page_size：覆盖 URL 参数
 – data：字典形式的过滤条件（键值对等值过滤）
 • 调用链：sysUserCrud.get_users(db, None, None, queryModel)。
 • 返回值：统一结构的Json。
总结
GET /users/{id} 用于“单条精确查”，GET /users/ 用于“无过滤分页列表”，POST /users/query/ 用于“带过滤条件、可自定义分页的列表”。

参数案例

1.精确查询ID资料：

类型：GET

/users/feddc9ed-8583-40c0-b3f1-9e2edbca6bd2

查询结果：

{"status_code": 200,"status": "success","message": "共计：1条数据","record_count": 1,"data": {"count": 1,"page": null,"page_size": null,"total_count": 1,"details": [{"is_superuser": false,"deletion_by": null,"is_staff": true,"create_date": "2025-08-20T16:39:09","id": "feddc9ed-8583-40c0-b3f1-9e2edbca6bd2","effective_date": "2025-04-24","creator_by": null,"expiry_date": null,"last_updated_date": null,"api_id": "some-api-id","deletion_mark": false,"last_updated_by": null,"login_password": null,"user_name": "some-user-name","deletion_reason": null,"login_id": "some-login-id4","email": "user@example.com","deletion_date": null,"person_id": "some-person-id"}]}
}

2.分页查询：

类型：GET

/users/?page=1&page_size=2

查询结果：

{"status_code": 200,"status": "success","message": "共计：2条数据","record_count": 2,"data": {"count": 2,"page": 1,"page_size": 2,"total_count": 31,"details": [{"is_superuser": false,"deletion_by": null,"is_staff": true,"create_date": "2025-08-20T14:56:25","id": "04714704-4fb6-4052-afde-9209b108d814","effective_date": "2025-04-24","creator_by": null,"expiry_date": null,"last_updated_date": null,"api_id": "some-api-id","deletion_mark": false,"last_updated_by": null,"login_password": "some-password","user_name": "some-user-name","deletion_reason": null,"login_id": "some-login-id","email": "user@example.com","deletion_date": null,"person_id": "some-person-id"},{"is_superuser": true,"deletion_by": null,"is_staff": true,"create_date": "2025-08-20T14:45:15","id": "10447332-d805-478f-a770-612aece26612","effective_date": "2023-03-01","creator_by": "admin","expiry_date": "2025-12-31","last_updated_date": "2023-03-01T00:00:00","api_id": "789","deletion_mark": false,"last_updated_by": "admin","login_password": "password789","user_name": "Charlie","deletion_reason": null,"login_id": "charlie","email": "charlie@example.com","deletion_date": null,"person_id": "P003"}]}
}

3.条件查询：

 类型：POST

/users/query/

查询参数Body资料

{"page": 2,"page_size": 2,"data": {"api_id": "some-api-id","login_id": "some-login-id"}
}

查询结果：

{"status_code": 200,"status": "success","message": "共计：2条数据","record_count": 2,"data": {"count": 2,"page": 2,"page_size": 2,"total_count": 12,"details": [{"is_superuser": false,"deletion_by": null,"is_staff": true,"create_date": "2025-08-20T14:14:41","id": "4a368e9b-306a-4123-8ce1-c27b07c0e1c6","effective_date": "2025-04-24","creator_by": null,"expiry_date": null,"last_updated_date": null,"last_updated_by": null,"api_id": "some-api-id","deletion_mark": false,"login_password": "some-password","user_name": "some-user-name","deletion_reason": null,"login_id": "some-login-id","email": "user@example.com","deletion_date": null,"person_id": "some-person-id"},{"is_superuser": false,"deletion_by": null,"is_staff": true,"create_date": "2025-08-20T14:48:26","id": "4a4b6673-ec46-4bf4-ab78-09fbe8616fc9","effective_date": "2025-04-24","creator_by": null,"expiry_date": null,"last_updated_date": null,"last_updated_by": null,"api_id": "some-api-id","deletion_mark": false,"login_password": "some-password","user_name": "some-user-name","deletion_reason": null,"login_id": "some-login-id","email": "user@example.com","deletion_date": null,"person_id": "some-person-id"}]}
}

封装

把查询代码进行封装，便于统一性；

在src\code新建orm_curd.py

from typing import Any, Dict, List, Optional, Type
from sqlalchemy.orm import Session, Query
from pydantic import BaseModel
from src.core import retMes, str_utils as strUdef query_with_page(db: Session,model_cls: Type,  # 对应 models.SysUserresp_model_cls: Optional[Type] = None,  # 对应 schemas.SysUser，仅做字段校验，可省略*,id: Optional[str] = None,filter_model: Optional[BaseModel] = None,  # 对应 request_model.QueryModelpage: int = 0,page_size: int = 100
) -> Dict[str, Any]:"""通用分页查询:param db: 数据库会话:param model_cls: SQLAlchemy 实体类:param resp_model_cls: Pydantic 响应模型（可选）:param id: 按主键 id 精确查询:param filter_model: 前端传来的过滤 + 分页参数:param page: 默认页码:param page_size: 默认每页条数:return: 按 retMes.Success 包装好的 dict"""query: Query = db.query(model_cls)# 1. 主键精确查询if strU.is_not_empty(id):query = query.filter(model_cls.id == id)# 2. 动态过滤条件elif strU.is_not_empty(filter_model):if filter_model.page is not None:page = filter_model.pageif filter_model.page_size is not None:page_size = filter_model.page_sizefilter_data: Dict[str, Any] = filter_model.data or {}for key, value in filter_data.items():if hasattr(model_cls, key) and value is not None:query = query.filter(getattr(model_cls, key) == value)# 3. 分页def _calc_skip(pg: int, ps: int) -> int:return (pg - 1) * ps + 1 if pg > 1 else 0if strU.is_empty(page) or strU.is_empty(page_size):total = query.count()rows = query.all()else:skip = _calc_skip(page, page_size)limit = page_sizetotal = query.count()rows = query.offset(skip).limit(limit).all()# 4. 转 dictdetails: List[Dict[str, Any]] = [strU.model_to_dict(r) for r in rows]# 5. 包装payload = {"count": len(details),"page": page,"page_size": page_size,"total_count": total,"details": details}return retMes.Success(payload,f'共计：{len(details)} 条数据',record_count=len(details)).mes()

修改src\curd\sys_user.py

from src.core import retMes, str_utils as strU, orm_curd
def get_users(db: Session, page: int = 0, page_size: int = 100, FilterModel: request_model.QueryModel = None,id: str = None):return orm_curd.query_with_page(db,models.SysUser,schemas.SysUser,  # 如不需要字段校验可传 Noneid=id,filter_model=FilterModel,page=page,page_size=page_size)

http://www.dtcms.com/a/353336.html

相关文章：

  • 【图文介绍】PCIe 6.0 Retimer板来了！
  • 快速上手对接币安加密货币API
  • 《Linux 网络编程四：TCP 并发服务器：构建模式、原理及关键技术（以select ）》
  • 3 无重复字符的最长子串
  • Windows系统之不使用第三方软件查看电脑详细配置信息
  • 基于linux系统的LIRC库学习笔记
  • Ubuntu 的磁盘管理
  • [java] 控制三个线程按顺序交替输出数字1、2、3
  • 【新版发布】Apache DolphinScheduler 3.3.1 正式上线：更稳、更快、更安全！
  • TensorFlow 面试题及详细答案 120道（21-30）-- 模型构建与神经网络
  • 数据结构：创建堆（或者叫“堆化”，Heapify）
  • 增强CD47检查点免疫治疗：高通量发现增强巨噬细胞吞噬作用的小分子协同剂
  • nestjs 连接redis
  • HIVE的Window functions窗口函数【一】
  • 手写题（面试）
  • LeetCode算法日记 - Day 24: 颜色分类、排序数组
  • LeetCode - 155. 最小栈
  • Python Imaging Library (PIL) 全面指南：PIL基础入门-跨平台安装与环境配置
  • Redis 数据结构
  • Linex系统网络管理（二）
  • 【yocto】Yocto Project 核心：深入了解.inc文件
  • Java中使用Spring Boot+Ollama构建本地对话机器人
  • Maven 依赖传递与排除基础逻辑
  • Astah UML 中，状态机（State Machine）的建模最合适使用「UML 状态图（State Diagram）」
  • 轻量级自动驾驶多视图视觉问答模型-EM-VLM4AD
  • 鸿蒙HarmonyOS状态管理装饰器详解
  • perccli 工具
  • 鸿蒙网络编程系列62-仓颉版使用Request部件上传多个文件到服务端
  • 华中科大联手小米推出ReCogDrive：自动驾驶迎来“认知革命”！
  • 零基础-力扣100题从易到难详解（持续更新）