一、项目背景

“后台订单列表翻到第 200 页时直接卡死了！”

星云电商运营后台的订单管理页面，支持按用户、日期范围、订单状态、支付方式、金额区间等 10 个维度的组合筛选。随着订单量突破 300 万，运营翻到第 200 页（LIMIT 50 OFFSET 10000）时页面白屏，数据库 CPU 飙升到 95%。

排查发现三个问题叠加导致了这次事故：

问题一：过滤条件用字符串拼接。20 个筛选条件有 20 个 if 分支，每个分支都在 SQL 字符串后面追了一段。因为 LIKE 和 IN 的拼接逻辑不同，一个引号处理失误导致 SQL 语法错误，整个页面报 500。

问题二：offset 分页在大偏移量下性能骤降。LIMIT 50 OFFSET 10000 要求数据库扫描前 10050 行再丢弃前 10000 行。当 offset 到 200 万时，数据库需要扫描 200 万行——即使有索引，这个开销也是 O(n) 的。

问题三：聚合查询未加缓存。运营每次打开页面，都会执行一个 COUNT(*) 计算总数，同时还要算多个金额的聚合（今日订单总额、本月订单总额、同比环比）。5 个聚合查询叠加，每次打开页面耗时 8 秒。

本章将围绕 SQLAlchemy 的过滤运算符（and_、or_、in_、like、between）、聚合函数（func.count/sum/avg）、分组（group_by/having）、子查询与 CTE 入门，以及两种分页模式（offset 分页 vs keyset 分页）展开实战。

二、项目设计

场景：周二下午，小胖和小白被拉进"后台慢查询治理"专项组。大师带来了白板和几杯咖啡。

大师：“先说过滤。回忆一下，如果用字符串拼 SQL，你的代码是什么样？”

小胖（苦笑）：

sql = "SELECT * FROM orders WHERE 1=1"
if user_id: sql += f" AND user_id = {user_id}"
if status:  sql += f" AND status = '{status}'"
if start:   sql += f" AND created_at >= '{start}'"
# ... 15 个条件后，已经不知道 AND 了几个了

大师：“而 SQLExpression 的写法：”

stmt = select(Order)
if user_id: stmt = stmt.where(Order.user_id == user_id)
if status:  stmt = stmt.where(Order.status == status)
if start:   stmt = stmt.where(Order.created_at >= start)

小胖：“技术映射：条件化拼装 = 搭积木，而不是拼接字符串。这两者的区别不仅是防注入——更重要的是可读性和复用性。”

小白：“那 and_ 和 or_ 什么时候需要显式用？我只是链式 .where()，它不就是 AND 吗？”

大师：“对。链式 .where() 就是 AND。但如果你的一个条件内部需要 OR，你就需要显式用 or_()：”

from sqlalchemy import or_, and_

# 查询：状态为 paid 且 (金额 > 1000 或 用户是 VIP)
stmt = select(Order).where(
    Order.status == "paid",
    or_(
        Order.total_amount > 1000,
        User.is_vip == True,
    )
)

小白：“技术映射：.where() 链式 = 自动 AND；or_() = 显式 OR。那 in_ 和 like 呢？”

大师：

# IN：状态是三个之一
stmt = stmt.where(Order.status.in_(["paid", "shipped", "completed"]))

# LIKE：模糊搜索（自动参数化，防注入）
stmt = stmt.where(Order.order_no.like("ORD-2024-%"))

# BETWEEN：日期范围
stmt = stmt.where(Order.created_at.between(start_date, end_date))

小胖：“那聚合查询呢？比如我要看每个用户的总消费额？”

大师：“用 func 聚合函数：”

from sqlalchemy import func

stmt = (
    select(
        Order.user_id,
        func.count(Order.id).label("order_count"),
        func.sum(Order.total_amount).label("total_spent"),
    )
    .group_by(Order.user_id)
    .having(func.sum(Order.total_amount) >= 1000)
    .order_by(func.sum(Order.total_amount).desc())
)

小胖：“这就像食堂的统计报表——各窗口（group_by）卖出多少份菜（count），总营业额多少（sum），只看营业额超 1000 的窗口（having）。”

大师：“技术映射：group_by = 分组统计维度；having = 分组后的过滤。”

小白：“终于到我最关心的分页了。offset 到底为什么慢？keyset 又是什么？”

大师：“offset 分页的工作原理是：数据库扫描前 N+M 行，丢掉前 N 行，返回 M 行。当 N=200 万时，数据库依然要扫描 200 万行。而 keyset（又称 seek 分页、游标分页）不依赖行号，而是用上一页最后一条记录的值作为下一页的起点。”

# offset 分页（慢在 N 大）
page_3 = select(Order).order_by(Order.id).limit(20).offset(40)

# keyset 分页（快，恒定为索引查找）
page_3 = select(Order).where(Order.id > last_id_of_page_2).order_by(Order.id).limit(20)

小胖：“技术映射：offset = 翻印刷书（一页页翻）；keyset = 书签跳转（直接翻到上次的位置）。”

小白：“那 keyset 有缺点吗？”

大师：“有。keyset 不能任意跳页——你不能一步跳到第 200 页，因为你不知道第 199 页最后一条的 ID 是什么。所以 keyset 适合’加载更多’（瀑布流、无限滚动），offset 适合’跳转到第 N 页’（但在 N 大时很慢）。实际生产中可以结合使用——前面几十页用 offset，后面用 keyset。”

三、项目实战

实战目标

实现订单后台列表的完整查询功能——多条件筛选（关键词、状态、日期、金额）、聚合统计（总金额、分类汇总）、以及 offset 和 keyset 两种分页的对比实现。

步骤一：模型与环境准备

"""ch12_filter_aggregate_paginate.py —— 过滤、聚合与分页实战"""

from sqlalchemy import (
    create_engine, String, Integer, Numeric, DateTime, Boolean,
    ForeignKey, text, func, select, update, and_, or_, not_,
    bindparam, literal, extract, cast,
)
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, relationship, sessionmaker, Session,
)
from datetime import datetime, timedelta
from typing import Optional, List
import random, uuid

engine = create_engine(
    "postgresql+psycopg://nebula:nebula_dev@localhost:5432/order_center",
    echo=False,
)

class Base(DeclarativeBase):
    pass

class Order(Base):
    __tablename__ = "fap_orders"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    order_no: Mapped[str] = mapped_column(String(32), unique=True, nullable=False, index=True)
    user_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True)
    user_name: Mapped[str] = mapped_column(String(50), nullable=False)
    total_amount: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False)
    status: Mapped[str] = mapped_column(String(20), nullable=False, index=True)
    payment_method: Mapped[str | None] = mapped_column(String(20), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True)
    paid_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    def __repr__(self):
        return f"<Order({self.order_no}, ¥{self.total_amount}, {self.status})>"

Base.metadata.create_all(engine)
SessionFactory = sessionmaker(bind=engine, autocommit=False, autoflush=False)

步骤二：插入测试数据

# =============================================
# 生成测试数据
# =============================================

statuses = ["pending", "paid", "shipped", "completed", "cancelled", "refunding"]
payments = ["wechat", "alipay", "card", None]

def generate_orders(n=500):
    """生成 n 条模拟订单"""
    base_time = datetime(2024, 1, 1, tzinfo=None)
    orders = []
    for i in range(1, n + 1):
        created = base_time + timedelta(days=random.randint(0, 200), hours=random.randint(0, 23))
        status = random.choices(
            statuses,
            weights=[0.05, 0.30, 0.25, 0.30, 0.05, 0.05], k=1
        )[0]
        orders.append({
            "order_no": f"ORD-{2024}-{i:06d}",
            "user_id": random.randint(1, 100),
            "user_name": f"user_{random.randint(1, 50)}",
            "total_amount": round(random.uniform(10, 2000), 2),
            "status": status,
            "payment_method": random.choice(payments),
            "created_at": created,
            "paid_at": created + timedelta(hours=random.randint(1, 24)) if status != "pending" else None,
        })
    return orders

with SessionFactory() as session:
    if session.execute(select(func.count(Order.id))).scalar() == 0:
        print("生成测试订单数据...")
        session.execute(Order.__table__.insert(), generate_orders(500))
        session.commit()
        print("已插入 500 条测试订单\n")

步骤三：多条件过滤查询

# =============================================
# 多条件筛选查询
# =============================================

def build_order_query(
    keyword: str = None,
    status: str = None,
    status_list: list[str] = None,
    user_name: str = None,
    min_amount: float = None,
    max_amount: float = None,
    start_date: datetime = None,
    end_date: datetime = None,
    payment_method: str = None,
    sort_by: str = "created_at_desc",
):
    """构建订单筛选查询——条件化拼装"""
    stmt = select(Order)

    # 关键词搜索——订单号 LIKE 或 用户名 LIKE
    if keyword:
        stmt = stmt.where(
            or_(
                Order.order_no.like(f"%{keyword}%"),
                Order.user_name.like(f"%{keyword}%"),
            )
        )

    # 精确状态（单选）
    if status:
        stmt = stmt.where(Order.status == status)

    # 多状态（多选，IN 查询）
    if status_list:
        stmt = stmt.where(Order.status.in_(status_list))

    # 用户名精确匹配
    if user_name:
        stmt = stmt.where(Order.user_name == user_name)

    # 金额区间（BETWEEN 用 >= 和 <= 实现）
    if min_amount is not None:
        stmt = stmt.where(Order.total_amount >= min_amount)
    if max_amount is not None:
        stmt = stmt.where(Order.total_amount <= max_amount)

    # 日期范围
    if start_date:
        stmt = stmt.where(Order.created_at >= start_date)
    if end_date:
        stmt = stmt.where(Order.created_at <= end_date)

    # 支付方式
    if payment_method:
        stmt = stmt.where(Order.payment_method == payment_method)

    # 排序
    sort_map = {
        "created_at_desc": Order.created_at.desc(),
        "created_at_asc": Order.created_at.asc(),
        "amount_desc": Order.total_amount.desc(),
        "amount_asc": Order.total_amount.asc(),
    }
    stmt = stmt.order_by(sort_map.get(sort_by, Order.created_at.desc()))

    return stmt

# 测试筛选查询
print("=== 多条件筛选查询 ===")
with SessionFactory() as session:
    # 筛选：已支付 + 金额 100-500 + 最近 30 天
    query = build_order_query(
        status="paid",
        min_amount=100,
        max_amount=500,
        start_date=datetime(2024, 6, 1),
        sort_by="amount_desc",
    )
    # 查看编译 SQL
    from sqlalchemy.dialects import postgresql
    compiled = query.compile(dialect=postgresql.dialect())
    print(f"SQL:\n{compiled}\n")

    results = session.execute(query.limit(10)).scalars().all()
    print(f"筛选结果（前 10 条，共 {len(results)} 条）:")
    for order in results:
        print(f"  {order.order_no} | {order.user_name} | ¥{order.total_amount} | {order.status} | {order.created_at.strftime('%Y-%m-%d')}")
print()

步骤四：聚合统计

# =============================================
# 聚合统计：分类汇总 + HAVING
# =============================================

print("=== 聚合统计 ===")

with SessionFactory() as session:
    # 统计一：按状态分组，查看各状态的订单数和总额
    print("--- 按状态统计 ---")
    stmt = (
        select(
            Order.status,
            func.count(Order.id).label("cnt"),
            func.sum(Order.total_amount).label("total"),
            func.round(func.avg(Order.total_amount), 2).label("avg"),
        )
        .group_by(Order.status)
        .having(func.count(Order.id) >= 1)
        .order_by(func.count(Order.id).desc())
    )
    print(f"{'状态':<15} {'订单数':<8} {'总额':<12} {'均价':<10}")
    print("-" * 45)
    for row in session.execute(stmt):
        print(f"{row.status:<15} {row.cnt:<8} ¥{row.total or 0:<11.2f} ¥{row.avg or 0:<9}")

    # 统计二：按日期聚合——最近 7 天每天的订单数
    print("\n--- 最近 7 天每日订单数 ---")
    seven_days_ago = datetime(2024, 7, 15)
    stmt = (
        select(
            func.date(Order.created_at).label("day"),
            func.count(Order.id).label("cnt"),
            func.sum(Order.total_amount).label("daily_total"),
        )
        .where(Order.created_at >= seven_days_ago)
        .group_by(func.date(Order.created_at))
        .order_by(func.date(Order.created_at))
    )
    for row in session.execute(stmt):
        print(f"  {row.day} | {row.cnt} 单 | ¥{row.daily_total:.2f}")

    # 统计三：大额用户——消费总额 > 5000 的用户
    print("\n--- 大额用户 (总消费 > 5000) ---")
    stmt = (
        select(
            Order.user_name,
            func.count(Order.id).label("order_count"),
            func.sum(Order.total_amount).label("total_spent"),
        )
        .group_by(Order.user_name)
        .having(func.sum(Order.total_amount) >= 5000)
        .order_by(func.sum(Order.total_amount).desc())
    )
    for row in session.execute(stmt):
        print(f"  {row.user_name}: {row.order_count} 单, ¥{row.total_spent:.2f}")

    # 统计四：子查询——消费额最高的用户详情
    print("\n--- 子查询：消费最高用户的订单明细 ---")
    subq = (
        select(
            Order.user_name,
            func.sum(Order.total_amount).label("total_spent"),
        )
        .group_by(Order.user_name)
        .order_by(func.sum(Order.total_amount).desc())
        .limit(1)
    ).cte("top_user")  # CTE 比子查询更高效

    top_orders = session.execute(
        select(Order)
        .where(Order.user_name == select(subq.c.user_name).scalar_subquery())
        .order_by(Order.created_at.desc())
        .limit(5)
    ).scalars().all()
    for o in top_orders:
        print(f"  {o.order_no} ¥{o.total_amount}")
print()

步骤五：两种分页对比

# =============================================
# offset 分页 vs keyset 分页
# =============================================

print("=== 分页对比：offset vs keyset ===")

with SessionFactory() as session:
    base_stmt = select(Order).order_by(Order.id)

    # 方式一：offset 分页（传统模式）
    print("--- offset 分页 ---")
    page = 3
    page_size = 10
    offset_stmt = base_stmt.limit(page_size).offset((page - 1) * page_size)
    page_3 = session.execute(offset_stmt).scalars().all()
    print(f"第 {page} 页: ids={[o.id for o in page_3]}")
    print(f"SQL: LIMIT {page_size} OFFSET {(page-1)*page_size}")

    # 方式二：keyset 分页（游标模式）
    print("\n--- keyset 分页 ---")
    # 获取第一页
    page_1 = session.execute(base_stmt.limit(page_size)).scalars().all()
    last_id_page_1 = page_1[-1].id
    print(f"第 1 页: ids={[o.id for o in page_1]}")
    print(f"最后 ID: {last_id_page_1}")

    # 用第一页最后一条的 ID 获取第二页
    page_2 = session.execute(
        base_stmt.where(Order.id > last_id_page_1).limit(page_size)
    ).scalars().all()
    print(f"第 2 页: ids={[o.id for o in page_2]}")
    print(f"SQL: WHERE id > {last_id_page_1} LIMIT {page_size}")

    # 查总数（用于分页导航）
    total = session.execute(
        select(func.count()).select_from(Order)
    ).scalar()
    print(f"\n总订单数: {total}, 总页数: {(total + page_size - 1) // page_size}")

# =============================================
# 通用 keyset 分页函数
# =============================================

def keyset_paginate(
    session, model, order_column, page_size: int = 20,
    cursor_value=None, filters: list = None, desc: bool = False,
):
    """
    通用 keyset 分页器
    - cursor_value: 上一页最后一条记录的排序字段值
    - filters: 额外的过滤条件列表
    """
    sort_col = order_column.desc() if desc else order_column.asc()
    stmt = select(model).order_by(sort_col).limit(page_size)

    if cursor_value is not None:
        if desc:
            # 降序：取比游标值小的
            stmt = stmt.where(order_column < cursor_value)
        else:
            # 升序：取比游标值大的
            stmt = stmt.where(order_column > cursor_value)

    if filters:
        for f in filters:
            stmt = stmt.where(f)

    result = session.execute(stmt).scalars().all()
    next_cursor = getattr(result[-1], order_column.name) if result else None
    return result, next_cursor

print("\n--- keyset 函数封装使用 ---")
with SessionFactory() as session:
    # 第一页
    page1, cursor1 = keyset_paginate(
        session, Order, Order.id, page_size=10,
        filters=[Order.status == "paid"],
    )
    print(f"第 1 页: {[o.id for o in page1]}, cursor={cursor1}")

    # 第二页
    page2, cursor2 = keyset_paginate(
        session, Order, Order.id, page_size=10,
        cursor_value=cursor1,
        filters=[Order.status == "paid"],
    )
    print(f"第 2 页: {[o.id for o in page2]}, cursor={cursor2}")

完整代码清单

"""ch12_filter_paginate_complete.py —— 过滤、聚合与分页完整示例"""

from sqlalchemy import (
    create_engine, String, Integer, Numeric, DateTime, ForeignKey,
    text, func, select, and_, or_, extract,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
from datetime import datetime
from typing import Optional, List
from order_center.config import DATABASE_URL

engine = create_engine(DATABASE_URL, echo=False)

class Base(DeclarativeBase):
    pass

class Order(Base):
    __tablename__ = "fp_orders"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    order_no: Mapped[str] = mapped_column(String(32), unique=True, nullable=False)
    user_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True)
    user_name: Mapped[str] = mapped_column(String(50), nullable=False)
    total_amount: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False)
    status: Mapped[str] = mapped_column(String(20), nullable=False, index=True)
    payment_method: Mapped[Optional[str]] = mapped_column(String(20))
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True)

Base.metadata.create_all(engine)
Factory = sessionmaker(bind=engine)

class OrderQuery:
    """订单查询封装"""
    def __init__(self, session):
        self.session = session

    def search(self, **kwargs) -> list[Order]:
        stmt = select(Order)
        if keyword := kwargs.get("keyword"):
            stmt = stmt.where(or_(Order.order_no.like(f"%{keyword}%"), Order.user_name.like(f"%{keyword}%")))
        if status := kwargs.get("status"):
            stmt = stmt.where(Order.status == status)
        if status_list := kwargs.get("status_list"):
            stmt = stmt.where(Order.status.in_(status_list))
        if min_amt := kwargs.get("min_amount"):
            stmt = stmt.where(Order.total_amount >= min_amt)
        if max_amt := kwargs.get("max_amount"):
            stmt = stmt.where(Order.total_amount <= max_amt)
        stmt = stmt.order_by(Order.created_at.desc())
        if limit := kwargs.get("limit"):
            stmt = stmt.limit(limit)
        if offset := kwargs.get("offset"):
            stmt = stmt.offset(offset)
        return list(self.session.execute(stmt).scalars().all())

    def stats_by_status(self) -> list:
        return self.session.execute(
            select(Order.status, func.count().label("cnt"), func.sum(Order.total_amount).label("total"))
            .group_by(Order.status).order_by(func.count().desc())
        ).all()

    def keyset_page(self, cursor_id: int = None, page_size: int = 20, **filters) -> tuple:
        stmt = select(Order).order_by(Order.id.asc()).limit(page_size)
        if cursor_id:
            stmt = stmt.where(Order.id > cursor_id)
        for k, v in filters.items():
            stmt = stmt.where(getattr(Order, k) == v)
        rows = list(self.session.execute(stmt).scalars().all())
        next_cursor = rows[-1].id if rows else None
        return rows, next_cursor

if __name__ == "__main__":
    print("OrderQuery 模块就绪")

可能遇到的坑及解决方法

1. and_() 和 or_() 条件嵌套优先级

• 现象：where(A or B and C) 实际生成了 WHERE A OR (B AND C)，而不是 WHERE (A OR B) AND C。
• 解决：显式用括号：or_(A, and_(B, C)) 或 and_(or_(A, B), C)，根据业务需求决定。

2. having 条件中不能用 Python 局部变量

• 现象：having(func.sum(Order.total_amount) >= min_value) 报错。
• 解决：用 bindparam() 或直接传字面量。having(func.sum(Order.total_amount) >= 1000)。

3. keyset 分页的排序字段必须是唯一的

• 现象：按 created_at 分页时漏数据（同一个时间戳有多条记录）。
• 根因：keyset 依赖排序字段的值作为游标，如果排序字段有重复值，> 可能跳过多条。
• 解决：排序字段加第二排序键（如 id）：order_by(Order.created_at.desc(), Order.id.asc())。

4. func 函数名拼写错误不报编译错误

• 现象：func.cout(...) 拼写错误，运行时才报 AttributeError: 'Function' object has no attribute 'cout'。
• 解决：IDE 没有 func 的自动补全，写测试验证 SQL 生成正确。

测试验证

# tests/test_ch12_filter.py
import pytest
from sqlalchemy import create_engine, String, Integer, Numeric, DateTime, text, func, select, or_, and_
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
from datetime import datetime

@pytest.fixture
def engine():
    return create_engine("sqlite:///:memory:", echo=False)

@pytest.fixture
def OrderModel(engine):
    class Base(DeclarativeBase):
        pass
    class Order(Base):
        __tablename__ = "orders"
        id: Mapped[int] = mapped_column(primary_key=True)
        order_no: Mapped[str] = mapped_column(String(32))
        user_name: Mapped[str] = mapped_column(String(50))
        total_amount: Mapped[float] = mapped_column(Numeric(12, 2))
        status: Mapped[str] = mapped_column(String(20))
        created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
    Base.metadata.create_all(engine)
    return Order

def test_where_chain_and(engine, OrderModel):
    """验证链式 where 为 AND"""
    Factory = sessionmaker(bind=engine)
    with Factory() as s:
        s.add_all([
            OrderModel(order_no="A", user_name="u1", total_amount=100, status="paid"),
            OrderModel(order_no="B", user_name="u2", total_amount=200, status="paid"),
            OrderModel(order_no="C", user_name="u1", total_amount=300, status="pending"),
        ])
        s.commit()

    with Factory() as s:
        stmt = select(OrderModel).where(OrderModel.status == "paid").where(OrderModel.total_amount >= 150)
        rows = s.execute(stmt).scalars().all()
        assert len(rows) == 1
        assert rows[0].order_no == "B"

def test_where_in_list(engine, OrderModel):
    """验证 IN 查询"""
    Factory = sessionmaker(bind=engine)
    with Factory() as s:
        s.add_all([
            OrderModel(order_no="X", user_name="u1", total_amount=10, status="paid"),
            OrderModel(order_no="Y", user_name="u1", total_amount=20, status="shipped"),
            OrderModel(order_no="Z", user_name="u1", total_amount=30, status="cancelled"),
        ])
        s.commit()
    with Factory() as s:
        stmt = select(OrderModel).where(OrderModel.status.in_(["paid", "shipped"]))
        rows = s.execute(stmt).scalars().all()
        assert len(rows) == 2

def test_group_by_having(engine, OrderModel):
    """验证分组聚合 + HAVING"""
    Factory = sessionmaker(bind=engine)
    with Factory() as s:
        s.add_all([
            OrderModel(order_no="a", user_name="u1", total_amount=100, status="paid"),
            OrderModel(order_no="b", user_name="u1", total_amount=200, status="paid"),
            OrderModel(order_no="c", user_name="u2", total_amount=50, status="paid"),
        ])
        s.commit()
    with Factory() as s:
        stmt = (
            select(OrderModel.user_name, func.sum(OrderModel.total_amount).label("total"))
            .group_by(OrderModel.user_name)
            .having(func.sum(OrderModel.total_amount) >= 200)
        )
        rows = s.execute(stmt).all()
        assert len(rows) == 1
        assert rows[0].user_name == "u1"

def test_keyset_pagination(engine, OrderModel):
    """验证 keyset 分页"""
    Factory = sessionmaker(bind=engine)
    with Factory() as s:
        for i in range(1, 26):
            s.add(OrderModel(order_no=f"NO-{i:03d}", user_name=f"u{i}", total_amount=i*10, status="paid"))
        s.commit()

    with Factory() as s:
        # 第一页（10 条）
        base = select(OrderModel).order_by(OrderModel.id.asc())
        p1 = s.execute(base.limit(10)).scalars().all()
        assert len(p1) == 10
        last_id = p1[-1].id
        # 第二页（keyset）
        p2 = s.execute(base.where(OrderModel.id > last_id).limit(10)).scalars().all()
        assert len(p2) == 10
        assert p2[0].id == 11

四、项目总结

优点与缺点

对比维度	字符串拼接 SQL	Expression 过滤/聚合/分页
条件化拼装	一长串 if 拼字符串	链式 .where() 对象组合
参数安全	手动占位符	自动参数化
聚合统计	手写 JOIN + GROUP BY	func.xxx + group_by 清晰表达
offsest 分页	手写 LIMIT/OFFSET	直接追加 .limit().offset()
keyset 分页	手写 WHERE id > last_id	同样的对象式构建

适用场景

1. 后台管理列表——多条件筛选 + 排序 + 分页。
2. 运营报表——按维度分组统计 + having 过滤。
3. 数据看板——聚合多个指标（今日/本周/本月）。
4. 移动端无限滚动——keyset 分页（“加载更多”）。
5. 导出功能——按条件筛选后流式导出。

不推荐场景：

1. 极其复杂的分析型查询（多层嵌套、窗口函数）——直接写原生 SQL 再用 text() 包裹。
2. 只有一两个固定条件的简单查询——用 session.get() 或简单 where 即可。

注意事项

1. having 和 where 的执行顺序不同：where 在分组前过滤行，having 在分组后过滤组。不要用 having 替代 where（性能差）。
2. keyset 分页需要排序字段上有索引：否则比 offset 更慢（每条都扫描）。
3. func.count() vs func.count(column)：COUNT(*) 包含 NULL，COUNT(column) 排除 NULL。用 func.count() 生成 COUNT(*)，用 func.count(Table.column) 生成 COUNT(column)。
4. 聚合结果在 Python 层可能是 Decimal 类型：Numeric(12, 2) 的聚合结果返回 Decimal 而非 float，JSON 序列化时需转换。

常见踩坑经验

案例 1：or_() 中嵌套了一个 and_() 但漏了括号

• 现象：筛选逻辑是 (A AND B) OR (C AND D)，但实际生成的是 A AND B OR C AND D（AND 优先级高于 OR，等价于 (A) AND (B OR C) AND (D)）。
• 修复：显式写 or_(and_(A, B), and_(C, D))。

案例 2：offset 很大的分页导致超时

• 现象：运营翻到第 500 页（offset 25000），数据库查询超时。
• 根因：MySQL/PostgreSQL 在 offset 大时必须扫描和丢弃大量行。
• 修复：限制最大翻页数（如最多 100 页），后续页改 keyset；或者用 ES 做搜索分离。

案例 3：func.string_agg 在 SQLite 和 PostgreSQL 上语法不同

• 现象：SQLite 测试通过，PostgreSQL 生产报 syntax error。
• 根因：string_agg(expr, delimiter) 在 SQLite 中第一个参数是表达式，PostgreSQL 中需要额外 ORDER BY。
• 修复：充分了解 dialect 差异，测试环境保持与生产数据库一致。

思考题

1. 你的订单系统需要支持"导出 100 万条订单到 CSV"功能。如果用一次 SELECT * FROM orders WHERE ... 全量加载到内存，会 OOM。请设计一个流式导出的方案，结合本章的过滤、排序和 keyset 分页技术，每次只加载 5000 条并追加到文件。

2. 运营需要一张"大额订单报表"：统计每个月订单金额超过 5000 元的订单数及占比。这个需求涉及分组（按月）、过滤（金额 > 5000）和子查询（求总订单数）。如果订单表有 500 万行，你如何优化这个查询？（提示：物化视图、分区表、summary 表）

---

参考答案参见附录 E。

延伸阅读与资源

NumPy 从入门到生产落地：全链路实战指南（科学计算/向量化）
Redis 8 实战精讲：从 CRUD 到源码，构建高可用缓存系统
Redis 实战修炼与原理进阶
Python 3实战精进：从脚本到高并发订单引擎
MongoDB 实战进阶与内核修炼
python入门：Rquests从菜鸟脚本到企业级SDK的网络实战圣经
Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地
后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战
10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析