一、为什么需要查询过滤与分页之前的 Repository 只有get_by_id和list_all——能查一条、能查全部。但真实业务从来不是给我所有记录用户只想看状态为 success 的记录搜索框输入Python要匹配 JD 原文里含这个词的记录首页只展示最近 7 天的分析结果数据库攒了 10000 条前端一次拉全部会卡死这四个场景分别对应过滤、搜索、时间范围、分页。list_all一个都解决不了。所以得给 Repository 加查询能力——这就是 3.1 要做的事。二、SQLAlchemy 查询三件套过滤、排序、分页核心机制Query 对象不可变SQLAlchemy 的session.query(Model)返回一个Query 对象它最大的特点是不可变——query.filter(...)不会修改原对象而是返回一个新的 Query。querysession.query(JdRecord)# 拿到原始 Queryquery.filter(JdRecord.statussuccess)# 这行白写没赋值回去recordsquery.all()# 查出来的还是全部记录必须重新赋值querysession.query(JdRecord)queryquery.filter(JdRecord.statussuccess)# 重新赋值才生效这个机制的好处是链式安全——你可以在条件分支里逐步叠加过滤条件不用担心互相污染。list_with_filter过滤 稳定排序 分页把过滤、排序、分页捏到一个方法里# app/db/repositories/jd_record_repo.pydeflist_with_filter(self,session:Session,status:str|NoneNone,keyword:str|NoneNone,limit:int20,offset:int0,)-list[JdRecord]:按状态/关键词筛选支持分页默认按创建时间倒序id 兜底稳定排序。querysession.query(JdRecord)ifstatusisnotNone:queryquery.filter(JdRecord.statusstatus)ifkeywordisnotNone:queryquery.filter(JdRecord.jd_text.like(f%{keyword}%))return(query.order_by(JdRecord.created_at.desc(),JdRecord.id.desc(),# id 兜底 → 稳定排序).offset(offset).limit(limit).all())两个关键设计稳定排序order_by(created_at.desc(), id.desc())如果多条记录的created_at完全相同批量插入时常见只按created_at排会出现同一页翻两次看到同一条或跳过某条。加id.desc()兜底保证排序结果唯一确定分页才不会乱。页码换算offset (page - 1) * sizepage 从 1 开始但 offset 从 0 开始。page1 → offset0第一条开始page2 → offset20跳过前 20 条。count 与 get_recent# app/db/repositories/jd_record_repo.pydefcount(self,session:Session,status:str|NoneNone)-int:统计符合过滤条件的记录总数statusNone 时统计全部。querysession.query(JdRecord)ifstatusisnotNone:queryquery.filter(JdRecord.statusstatus)returnquery.count()defget_recent(self,session:Session,days:int7)-list[JdRecord]:按创建时间倒序列出最近 N 天的记录。cutoffdatetime.now()-timedelta(daysdays)return(session.query(JdRecord).filter(JdRecord.created_atcutoff).order_by(JdRecord.created_at.desc(),JdRecord.id.desc()).all())count的过滤逻辑和list_with_filter的前半段完全一致——同一个if status is not None分支。区别只在最后一步一个.all()返回记录列表一个.count()返回数字。小结Query 对象不可变是 SQLAlchemy 查询的核心心智模型。记住filter 返回新对象必须赋值回去能避免 80% 的查询 bug。稳定排序是分页的隐形前提——没有稳定排序分页就会漏数据或重复数据。三、从数据库到 API为什么要分页列表接口3.1 解决了数据库怎么查3.2 解决怎么把查到的数据通过 API 给前端。如果直接return db.query(JdRecord).all()给前端数据库里有 10000 条记录会发生什么服务器1 万条记录一把查出来内存飙、响应耗时炸前端浏览器渲染 1 万行 DOM页面卡死用户面对 1 万条数据根本找不到想要的那条分页的本质是把大海捞针变成翻书阅读一次只看一页。接口设计请求GET /api/v1/jd/records?page1size20statussuccess响应不直接返回数组而是包一层{items:[{id:1,job_title:Python后端,status:success,created_at:...},{id:2,job_title:AI工程师,status:success,created_at:...}],total:100,page:1,size:20}为什么包一层而不是直接return [...]total是给前端算还有几页用的。前端拿到total100、size20就能算出总页数ceil(100/20)5画出分页导航条[上一页] 1 2 [3] 4 5 [下一页]。如果只返回数组前端不知道后面还有没有数据分页控件画不出来。四、泛型分页包装 ORM ≠ SchemaPaginatedResponse写一次管所有数据类型分页响应的结构items total page size是通用的——JD 列表要用将来用户列表、订单列表也要用。如果每种实体写一个XxxPaginatedResponse代码重复到崩溃。Python 的泛型Generic解决这个问题——写一个模具往里面填类型# app/schemas/common.pyfromtypingimportGeneric,TypeVar TTypeVar(T)classPaginatedResponse(BaseModel,Generic[T]):通用分页响应包装。T 在使用时替换为具体类型。items:list[T]total:intpage:intsize:intT TypeVar(T)声明一个占位符“到时候再说是什么类型”Generic[T]告诉 Python “这个类肚子里有个可变类型槽位”items: list[T]的意思items 装什么取决于 T 填什么使用时用方括号指定具体类型PaginatedResponse[JdRecordResponse]→ FastAPI 自动理解items是list[JdRecordResponse]生成正确的 API 文档。JdRecordResponse只暴露该暴露的字段# app/schemas/common.pyclassJdRecordResponse(BaseModel):id:intjob_title:str|Nonestatus:strcreated_at:datetime model_config{from_attributes:True}ORM 模型JdRecord有 9 个字段含jd_text、error_message、analysis_result等内部字段但 API 只给前端 4 个。这就是ORM 模型 ≠ Pydantic Schema的职责边界ORM 模型 (JdRecord)Pydantic Schema (JdRecordResponse)职责完整映射数据库表定义 API 输出的形状字段策略越全越好不能丢越少越好只给需要的面向谁后端代码增删改查前端 / API 消费者直接返回 ORM 对象 把后台更衣室全裸推到 T 台上——error_message、将来加的raw_api_response可能含敏感信息全跟着出去。model_config {from_attributes: True}让 Pydantic 能从 ORM 对象上按 Schema 定义的字段逐个取值不在 Schema 里的自动忽略。五、路由实现与测试路由Query 参数边界 ORM→Schema 转换# app/api/routes/jd.pyrouter.get(/records,response_modelPaginatedResponse[JdRecordResponse],summary获取 JD 分析历史记录列表,)deflist_records(page:intQuery(1,ge1,description页码从1开始),size:intQuery(20,ge1,le100,description每页条数1-100),status:str|NoneQuery(None,description按状态过滤: success/failed/pending),):withSessionLocal()asdb:recordsrepo.list_with_filter(db,statusstatus,limitsize,offset(page-1)*size)totalrepo.count(db,statusstatus)items[JdRecordResponse.model_validate(r)forrinrecords]returnPaginatedResponse[JdRecordResponse](itemsitems,totaltotal,pagepage,sizesize)三个设计要点Query(1, ge1)/Query(20, ge1, le100)给参数加边界。ge1大于等于1、le100小于等于100。用户传?page0或?size999→ FastAPI 自动返回 422不用手写 if 判断。with SessionLocal() as db创建 session用完自动关。这是 3.2 的过渡写法4.1 会升级为Depends(get_db)依赖注入。[JdRecordResponse.model_validate(r) for r in records]ORM 对象逐个转 Schema。model_validate配合from_attributesTrue从 ORM 对象上取 Schema 定义的那 4 个字段其余忽略。测试4 场景验证# tests/test_list_records.pypytest.fixture(autouseTrue)defclean_jd_records():每个测试前后清空 jd_records 表避免污染开发库。withSessionLocal()asdb:db.query(JdRecord).delete()db.commit()yieldwithSessionLocal()asdb:db.query(JdRecord).delete()db.commit()# 1. 空数据库 → items[]、total0deftest_list_records_empty_returns_empty(client:TestClient):responseclient.get(/api/v1/jd/records)assertresponse.status_code200dataresponse.json()assertdata[items][]assertdata[total]0# 2. 插入 3 条 → 返回 3 条deftest_list_records_with_three_records_returns_all(client:TestClient):_make_record(statussuccess)_make_record(statussuccess,job_titleAI 工程师)_make_record(statusfailed)responseclient.get(/api/v1/jd/records)assertdata[total]3assertlen(data[items])3# 3. 插入 25 条page2 size20 → 返回 5 条第 21-25deftest_list_records_pagination_page_2_size_20(client:TestClient):foriinrange(25):_make_record(job_titlefJD{i:02d})responseclient.get(/api/v1/jd/records?page2size20)assertdata[total]25assertlen(data[items])5# 4. 非法分页参数 → 422deftest_list_records_invalid_params_returns_422(client:TestClient):r1client.get(/api/v1/jd/records?page0)assertr1.status_code422r2client.get(/api/v1/jd/records?size999)assertr2.status_code422autouseTrue的 fixture 每个测试前后自动清表——这是 3.2 的临时方案4.1 会升级为内存数据库 依赖注入隔离连开发库都不碰。这两关把数据从能查升级到查得准、给得对——3.1 让 Repository 学会过滤/排序/分页3.2 让这些能力通过 API 有序地交给前端。核心就三件事Query 对象不可变filter 要赋值回去、稳定排序是分页的前提id 兜底、ORM 和 Schema 职责分离别把数据库模型直接丢给前端。泛型PaginatedResponse写一次管所有列表接口是这趟最值的复用。