一、项目背景

"这个 SQL 到底查了哪些字段？"小白盯着屏幕上一条 200 多行的 SQL 语句，头皮发麻。

星云电商订单中台有一个"订单综合查询"接口，支持按用户、时间范围、订单状态、支付方式、金额区间等 15 个维度的组合筛选。原始实现是用字符串拼接构造 SQL——每当用户选择了一个筛选条件，就在 SQL 字符串后面追加一段 WHERE 子句。代码大概长这样：

sql = "SELECT o.*, u.username FROM orders o JOIN users u ON o.user_id = u.id WHERE 1=1"
if user_name:
    sql += f" AND u.username = '{user_name}'"          # SQL 注入漏洞！
if start_date:
    sql += f" AND o.created_at >= '{start_date}'"      # 日期格式不统一会报错
if statuses:
    sql += f" AND o.status IN ({','.join(statuses)})"  # 列表拼接不处理引号
# ... 继续追加 10+ 个条件

上次安全审计时发现这里有高危 SQL 注入漏洞——user_name 参数直接拼接进 SQL，攻击者可以传入 ' OR 1=1 -- 绕过密码验证。更讽刺的是，尽管团队知道 SQL 注入的原理，但因为筛选维度太多且组合复杂，没人敢动这块代码。

另一个问题是在做分页列表时，LIMIT 100 OFFSET 2000 的方式在数据量大时性能极差——这是经典的 offset 分页陷阱（第12章详解）。

SQLAlchemy 的 SQL Expression Language 就是为解决这类问题设计的。它把 SQL 变成可组合的 Python 对象，既能防止注入，又能条件化拼接，还能方便地查看编译后的 SQL 语句和绑定参数。本章将以商品列表查询为核心场景，学习 select()、where、order_by、limit/offset、join 等核心表达式的用法，以及从 Core Result 到字典/命名元组的转换技巧。

二、项目设计

场景：周五上午，工位区。小胖正在写一个新的订单查询接口，但他没有像以前那样拼接字符串，而是用了一堆 Python 函数。

小胖：“大师，这个 select() 是怎么回事？我以前都直接写 SQL 字符串，现在要写一堆 select(users).where(...) 感觉反而更啰嗦了。这不是脱裤子放屁吗？”

大师：“那我问你：如果用字符串拼接，你怎么保证在 15 个可选筛选条件下不写出 SQL 注入漏洞？”

小胖：“呃……每个参数都用 %s 占位符？那代码会变成这样：”

conditions = []
params = []
if name:
    conditions.append("u.username = %s")
    params.append(name)
if start:
    conditions.append("o.created_at >= %s")
    params.append(start)
# ... 10 个条件后，我已经数不清 %s 的数量了

小胖：“而且一旦中间插入了一个新条件，所有 %s 的顺序都要检查。”

大师：“这就是 SQL Expression Language 的价值。它不是啰嗦，而是把 SQL 变成可组合对象，让 Python 帮你管理参数顺序和安全。你还记得食堂的自助餐吗？”

小胖：“又拿食堂打比方……这次是什么？”

大师：“原始的 SQL 就像你去食堂直接冲进后厨对着灶台炒菜——快但危险。SQL Expression 是用选菜盆——你把想要的菜（Column）装进盆（select），用夹子（where）夹走不想要的，用标签（order_by）排好顺序，最后一次性送进窗口（execute）。”

小胖：“技术映射：SQL Expression = 用 Python 对象组装 SQL 的选菜盆。有点意思。那实际写起来呢？select 是不是一开始就要定好查什么？”

大师：“select() 的核心优势就是可逐步构建。比如你写一个商品搜索接口：”

from sqlalchemy import select

# 基础查询
stmt = select(products)

# 条件化添加 WHERE
if keyword:
    stmt = stmt.where(products.c.title.ilike(f"%{keyword}%"))
if category:
    stmt = stmt.where(products.c.category == category)
if min_price is not None:
    stmt = stmt.where(products.c.unit_price >= min_price)

# 条件化添加排序
if sort_by == "price_asc":
    stmt = stmt.order_by(products.c.unit_price.asc())
elif sort_by == "price_desc":
    stmt = stmt.order_by(products.c.unit_price.desc())

# 分页
stmt = stmt.limit(page_size).offset((page - 1) * page_size)

小白：“这个写法确实比追加大段字符串干净。但有一个问题：我注意到你用了 products.c.title？这 c 是什么意思？”

大师：“c 代表 Columns——Table 对象的列集合。products.c.title 等价于你直接写 'products.title'，但它是一个 Python 对象，SQLAlchemy 可以自动处理它的类型、绑定参数和引用名。”

小白：“技术映射：Table.c = 列访问器，将列名转为类型安全的对象。那参数绑定是怎么防 SQL 注入的？如果我传 keyword = "'; DROP TABLE products; --"，它会怎么处理？”

大师：“来看编译结果：”

stmt = select(products).where(products.c.title.ilike(f"%'; DROP TABLE products; --%"))
print(stmt.compile(compile_kwargs={"literal_binds": True}))
# 输出: SELECT products.id, ... FROM products
#        WHERE products.title ILIKE '%'; DROP TABLE products; --%'

大师：“等等……这不是拼进去了吗？”

（小胖和小白同时倒吸一口凉气）

大师：“哈哈，别慌——这只是 compile(literal_binds=True) 的可视化输出！实际上发给数据库的是参数化查询。”

# 实际的参数化查询（默认行为）
compiled = stmt.compile()
print(compiled)    # WHERE products.title ILIKE %(title_1)s
print(compiled.params)  # {'title_1': "'; DROP TABLE products; --"}

小胖：“哦！所以真正传给数据库的是两个独立的东西：SQL 模板和参数值。参数值里的 SQL 关键字永远不会被解析为 SQL 语法。”

大师：“技术映射：参数化查询 = SQL 模板 + 参数值双通道发送，值被隔离为纯数据。这下放心了？”

小白：“那 JOIN 呢？原生的 JOIN 写起来很繁琐。SQLAlchemy 的 JOIN 怎么弄？”

大师：“JOIN 也是对象！”

# 假设有 users 和 orders 两张表
stmt = (
    select(
        orders.c.id,
        orders.c.order_no,
        users.c.username,
        orders.c.total_amount,
    )
    .select_from(orders.join(users, orders.c.user_id == users.c.id))
    .where(users.c.is_active == True)
    .order_by(orders.c.created_at.desc())
)

小胖：“也就是说，JOIN 也是通过 .join() 方法追加的？和 WHERE 一样，条件化拼装？”

大师：“完全正确。如果你不确定某个表是否需要 JOIN（取决于筛选条件），你可以先不 join，只在条件满足时追加 .select_from(orders.join(users, ...))。”

小白：“那返回的结果我是怎么拿到的？也是对象吗？”

大师：“Core 执行返回的是 Row 对象，支持多种消费方式：”

with engine.connect() as conn:
    result = conn.execute(stmt)

    # 逐行遍历（大结果集推荐）
    for row in result:
        print(f"{row.id} | {row.order_no} | {row.username}")

    # 取所有行
    rows = result.all()  # List[Row]

    # 转为字典列表（适合返回 JSON）
    dicts = [dict(row._mapping) for row in rows]

小胖：“技术映射：Row._mapping = 行到字典的桥梁。那 label 和 alias 在什么场景用？”

大师：“label 用于给列起别名，alias 用于给子查询起别名。”

from sqlalchemy import func, alias

# label: 给聚合函数结果起一个可读的名字
stmt = select(
    func.count(products.c.id).label("total"),
    products.c.category,
).group_by(products.c.category)

# alias: 在自连接或子查询中给表/查询起别名
subquery = (
    select(orders.c.user_id, func.count(orders.c.id).label("order_count"))
    .group_by(orders.c.user_id)
).alias("order_stats")

stmt = select(users, subquery.c.order_count).select_from(
    users.join(subquery, users.c.id == subquery.c.user_id)
)

三、项目实战

实战目标

以"星云电商商品管理后台"为场景，实现商品列表查询的完整功能：多条件筛选（关键词、分类、价格区间、状态）、多字段排序、分页；打印编译后的 SQL 和绑定参数；对比 offset 分页和 keyset 分页的 SQL 差异。

步骤一：建立基础 Table 与插入测试数据

"""ch05_expression_select.py —— 商品查询的 SQL Expression 实战"""

from sqlalchemy import (
    MetaData, Table, Column, Integer, String, Text,
    Numeric, DateTime, Boolean, JSON, Index, CheckConstraint,
    create_engine, select, and_, or_, not_, func, text,
    bindparam,
)
from sqlalchemy.sql import func as sqlfunc

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

metadata = MetaData()

products = Table(
    "products", metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("sku", String(30), unique=True, nullable=False),
    Column("title", String(200), nullable=False),
    Column("description", Text),
    Column("unit_price", Numeric(12, 2), nullable=False),
    Column("inventory", Integer, nullable=False, server_default=text("0")),
    Column("category", String(50), nullable=False, index=True),
    Column("tags", JSON),
    Column("status", String(20), nullable=False, server_default=text("'online'")),
    Column("is_hot", Boolean, server_default=text("FALSE")),
    Column("created_at", DateTime(timezone=True), server_default=sqlfunc.now()),
    CheckConstraint("unit_price > 0", name="ck_price_positive"),
    CheckConstraint("inventory >= 0", name="ck_inventory_non_negative"),
)

metadata.create_all(engine)

# 插入测试数据
test_products = [
    {"sku": "SKU-001", "title": "有机全麦面包", "unit_price": 15.80, "inventory": 500, "category": "食品", "tags": ["有机", "烘焙"], "status": "online", "is_hot": True},
    {"sku": "SKU-002", "title": "低脂酸奶 100g", "unit_price": 8.50, "inventory": 320, "category": "食品", "tags": ["低脂", "乳制品"], "status": "online", "is_hot": False},
    {"sku": "SKU-003", "title": "无线蓝牙耳机 Pro", "unit_price": 299.00, "inventory": 80, "category": "电子产品", "tags": ["蓝牙", "音频"], "status": "online", "is_hot": True},
    {"sku": "SKU-004", "title": "游戏机械键盘 87键", "unit_price": 459.00, "inventory": 45, "category": "电子产品", "tags": ["键盘", "游戏"], "status": "online", "is_hot": True},
    {"sku": "SKU-005", "title": "棉质T恤 白色", "unit_price": 79.00, "inventory": 1200, "category": "服装", "tags": ["纯棉", "基础款"], "status": "online", "is_hot": False},
    {"sku": "SKU-006", "title": "防晒霜 SPF50", "unit_price": 129.00, "inventory": 0, "category": "护肤品", "tags": ["防晒", "户外"], "status": "offline", "is_hot": False},
    {"sku": "SKU-007", "title": "有机绿茶礼盒", "unit_price": 238.00, "inventory": 15, "category": "食品", "tags": ["有机", "茶叶", "礼盒"], "status": "online", "is_hot": False},
    {"sku": "SKU-008", "title": "智能手表 S3", "unit_price": 1899.00, "inventory": 30, "category": "电子产品", "tags": ["穿戴", "智能"], "status": "online", "is_hot": True},
    {"sku": "SKU-009", "title": "碳钢不粘锅 28cm", "unit_price": 299.00, "inventory": 200, "category": "厨具", "tags": ["不粘", "碳钢"], "status": "online", "is_hot": False},
    {"sku": "SKU-010", "title": "瑜伽垫 6mm", "unit_price": 49.00, "inventory": 800, "category": "运动", "tags": ["瑜伽", "健身"], "status": "online", "is_hot": False},
]

with engine.connect() as conn:
    with conn.begin():
        conn.execute(products.insert(), test_products)
    print(f"已插入 {len(test_products)} 条测试数据")

步骤二：多条件筛选查询

# =============================================
# 步骤二：实现商品搜索 —— 多条件 + 排序 + 分页
# =============================================

def search_products(
    keyword: str = None,
    category: str = None,
    min_price: float = None,
    max_price: float = None,
    status: str = "online",
    sort_by: str = "created_at_desc",
    page: int = 1,
    page_size: int = 10,
):
    """通用商品搜索 —— 使用 SQL Expression 条件化拼装 SQL"""

    # 基础查询
    stmt = select(products).where(products.c.status == status)

    # 关键词搜索（模糊匹配标题）
    if keyword:
        stmt = stmt.where(products.c.title.ilike(f"%{keyword}%"))

    # 分类筛选
    if category:
        stmt = stmt.where(products.c.category == category)

    # 价格区间（使用 and_ 组合多个条件）
    price_conditions = []
    if min_price is not None:
        price_conditions.append(products.c.unit_price >= min_price)
    if max_price is not None:
        price_conditions.append(products.c.unit_price <= max_price)
    if price_conditions:
        stmt = stmt.where(and_(*price_conditions))

    # 排序（根据传入参数动态切换）
    sort_fns = {
        "created_at_desc": products.c.created_at.desc(),
        "created_at_asc": products.c.created_at.asc(),
        "price_asc": products.c.unit_price.asc(),
        "price_desc": products.c.unit_price.desc(),
        "inventory_asc": products.c.inventory.asc(),
    }
    if sort_by in sort_fns:
        stmt = stmt.order_by(sort_fns[sort_by])

    # 分页
    stmt = stmt.limit(page_size).offset((page - 1) * page_size)

    return stmt

# 测试查询
print("\n=== 测试查询 1：搜索'蓝牙'，电子类，价格 100-500，按价格升序 ===")
stmt1 = search_products(
    keyword="蓝牙",
    category="电子产品",
    min_price=100,
    max_price=500,
    sort_by="price_asc",
)

# 查看编译后的 SQL
compiled = stmt1.compile(compile_kwargs={"literal_binds": True})
print(f"SQL:\n{compiled}")
print(f"\n参数化参数: {stmt1.compile().params}")

with engine.connect() as conn:
    result = conn.execute(stmt1)
    print(f"\n查询结果（共 {result.rowcount} 行）:")
    for row in result:
        print(f"  [{row.category}] {row.title} - ¥{row.unit_price} (库存 {row.inventory})")

运行结果：

=== 测试查询 1：搜索'蓝牙'，电子类，价格 100-500，按价格升序 ===
SQL:
SELECT products.id, products.sku, products.title, products.description,
products.unit_price, products.inventory, products.category, products.tags,
products.status, products.is_hot, products.created_at
FROM products
WHERE products.status = 'online'
  AND products.title ILIKE '%蓝牙%'
  AND products.category = '电子产品'
  AND products.unit_price >= 100.0
  AND products.unit_price <= 500.0
ORDER BY products.unit_price ASC
LIMIT 10 OFFSET 0

查询结果（共 1 行）:
  [电子产品] 无线蓝牙耳机 Pro - ¥299.00 (库存 80)

步骤三：聚合统计与分组

# =============================================
# 步骤三：分类汇总统计
# =============================================

print("\n=== 各分类商品统计数据 ===")

# 使用 func 聚合函数 + label 命名
stats_stmt = (
    select(
        products.c.category,
        func.count(products.c.id).label("total"),
        func.round(func.avg(products.c.unit_price), 2).label("avg_price"),
        func.sum(products.c.inventory).label("total_inventory"),
        func.sum(products.c.is_hot.cast(Integer)).label("hot_count"),
    )
    .where(products.c.status == "online")
    .group_by(products.c.category)
    .having(func.count(products.c.id) >= 1)
    .order_by(func.count(products.c.id).desc())
)

compiled = stats_stmt.compile(compile_kwargs={"literal_binds": True})
print(f"SQL:\n{compiled}")

with engine.connect() as conn:
    result = conn.execute(stats_stmt)
    print(f"\n{'分类':<10} {'商品数':<8} {'均价':<10} {'总库存':<10} {'热门数':<8}")
    print("-" * 50)
    for row in result:
        print(f"{row.category:<10} {row.total:<8} ¥{row.avg_price:<9} {row.total_inventory:<10} {row.hot_count:<8}")

运行结果：

SQL:
SELECT products.category,
       count(products.id) AS total,
       round(avg(products.unit_price), 2) AS avg_price,
       sum(products.inventory) AS total_inventory,
       sum(CAST(products.is_hot AS INTEGER)) AS hot_count
FROM products
WHERE products.status = 'online'
GROUP BY products.category
HAVING count(products.id) >= 1
ORDER BY count(products.id) DESC

分类       商品数   均价       总库存     热门数
--------------------------------------------------
电子产品   3       ¥885.67    155       3
食品       2       ¥127.15    835       0
服装       1       ¥79.00     1200      0
护肤品     0       0          0         0
厨具       1       ¥299.00    200       0
运动       1       ¥49.00     800       0

步骤四：JOIN 查询演示

# =============================================
# 步骤四：JOIN 查询 —— 商品 + 库存日志（假设表）
# =============================================

# 先创建一张关联表
inventory_logs = Table(
    "inventory_logs", metadata,
    Column("id", Integer, primary_key=True),
    Column("product_id", Integer, nullable=False),
    Column("change_amount", Integer, nullable=False),
    Column("change_type", String(20), nullable=False),  # 'in'/'out'
    Column("created_at", DateTime(timezone=True), server_default=sqlfunc.now()),
)
metadata.create_all(engine)

# 插入测试日志
with engine.connect() as conn:
    with conn.begin():
        conn.execute(inventory_logs.insert(), [
            {"product_id": 1, "change_amount": 500, "change_type": "in"},
            {"product_id": 1, "change_amount": -10, "change_type": "out"},
            {"product_id": 3, "change_amount": -2, "change_type": "out"},
            {"product_id": 3, "change_amount": 80, "change_type": "in"},
        ])

print("\n=== JOIN: 商品库存变动明细 ===")
join_stmt = (
    select(
        products.c.sku,
        products.c.title,
        inventory_logs.c.change_type,
        inventory_logs.c.change_amount,
        inventory_logs.c.created_at,
    )
    .select_from(
        products.join(inventory_logs, products.c.id == inventory_logs.c.product_id)
    )
    .order_by(products.c.id, inventory_logs.c.created_at)
)

compiled = join_stmt.compile(compile_kwargs={"literal_binds": True})
print(f"SQL:\n{compiled}")

with engine.connect() as conn:
    for row in conn.execute(join_stmt):
        direction = "↑" if row.change_amount > 0 else "↓"
        print(f"  [{row.sku}] {row.title:20s} {direction}{abs(row.change_amount):>4}")

步骤五：标准分页查询与 count 查询

# =============================================
# 步骤五：分页查询模板（列表页标准写法）
# =============================================

def paginated_query(stmt, page=1, page_size=10):
    """统一的分页查询包装器"""
    count_stmt = (
        select(func.count())
        .select_from(stmt.subquery())  # 在外层包一层 count
    )
    page_stmt = stmt.limit(page_size).offset((page - 1) * page_size)
    return count_stmt, page_stmt

# 使用示例
base_stmt = (
    select(products.c.id, products.c.sku, products.c.title, products.c.unit_price)
    .where(products.c.status == "online")
    .order_by(products.c.created_at.desc())
)

count_sql, page_sql = paginated_query(base_stmt, page=1, page_size=5)

with engine.connect() as conn:
    total = conn.execute(count_sql).scalar()
    rows = conn.execute(page_sql)
    print(f"\n=== 分页查询（第1页，每页5条，共 {total} 条）===")
    for row in rows:
        print(f"  {row.sku:10s} {row.title:25s} ¥{row.unit_price}")

完整代码清单

"""ch05_expression_complete.py —— SQL Expression Language 完整示例"""

from sqlalchemy import (
    MetaData, Table, Column, Integer, String, Text, Numeric,
    DateTime, Boolean, JSON, Index, CheckConstraint,
    create_engine, select, and_, or_, func, text, bindparam,
)
from sqlalchemy.sql import func as sqlfunc
from order_center.config import DATABASE_URL

engine = create_engine(DATABASE_URL, echo=False)
metadata = MetaData()

products = Table(
    "products", metadata,
    Column("id", Integer, primary_key=True),
    Column("sku", String(30), unique=True, nullable=False),
    Column("title", String(200), nullable=False),
    Column("unit_price", Numeric(12, 2), nullable=False),
    Column("inventory", Integer, nullable=False, server_default=text("0")),
    Column("category", String(50), nullable=False, index=True),
    Column("status", String(20), nullable=False, server_default=text("'online'")),
    Column("created_at", DateTime(timezone=True), server_default=sqlfunc.now()),
    CheckConstraint("unit_price > 0"),
    CheckConstraint("inventory >= 0"),
)

# ========== 查询执行 ==========
metadata.create_all(engine)

def build_search_query(keyword=None, category=None, min_price=None, max_price=None,
                       sort_by="created_at_desc", page=1, page_size=10):
    stmt = select(products).where(products.c.status == "online")
    if keyword:
        stmt = stmt.where(products.c.title.ilike(f"%{keyword}%"))
    if category:
        stmt = stmt.where(products.c.category == category)
    if min_price is not None:
        stmt = stmt.where(products.c.unit_price >= min_price)
    if max_price is not None:
        stmt = stmt.where(products.c.unit_price <= max_price)
    sort_map = {
        "created_at_desc": products.c.created_at.desc(),
        "price_asc": products.c.unit_price.asc(),
    }
    stmt = stmt.order_by(sort_map.get(sort_by, products.c.created_at.desc()))
    stmt = stmt.limit(page_size).offset((page - 1) * page_size)
    return stmt

if __name__ == "__main__":
    stmt = build_search_query(category="电子产品", min_price=100)
    compiled = stmt.compile(compile_kwargs={"literal_binds": True})
    print(compiled)
    print(f"参数: {stmt.compile().params}")

    with engine.connect() as conn:
        for row in conn.execute(stmt):
            print(f"  {row.sku}: {row.title} @ ¥{row.unit_price}")

可能遇到的坑及解决方法

1. where 多次调用是 AND 叠加，不是替换

• 现象：连续调用 .where() 三次，期望覆盖前面的条件，结果生成了三个 AND。
• 原因：select.where() 是把新条件 AND 到已有条件上，不是替换。
• 解决：如果条件可变，把公共条件放在函数末尾的 .where() 中。

2. ilike 需要手动加 % 通配符

• 现象：products.c.title.ilike("蓝牙") 查不到数据，但 products.c.title == "蓝牙耳机" 能查到。
• 原因：ilike 默认是精确匹配（忽略大小写），不等于 LIKE 的包含匹配。
• 解决：显式加通配符 f"%{keyword}%"。

3. compile(literal_binds=True) 显示了拼接值，并不意味着实际执行就是拼接的

• 现象：开发者用 compile 看到输出中嵌入了参数值，误以为有 SQL 注入风险。
• 原因：literal_binds=True 是为了方便阅读的命令行输出，实际执行仍然是参数化查询。
• 解决：理解 compile() 默认行为和 literal_binds=True 的区别。

测试验证

# tests/test_ch05_expression.py
import pytest
from sqlalchemy import (
    MetaData, Table, Column, Integer, String, Numeric,
    create_engine, select, func, text, CheckConstraint,
)
from sqlalchemy.sql import func as sqlfunc

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

@pytest.fixture
def products_table(engine):
    meta = MetaData()
    products = Table(
        "products", meta,
        Column("id", Integer, primary_key=True),
        Column("title", String(200), nullable=False),
        Column("unit_price", Numeric(12, 2), nullable=False),
        Column("category", String(50), nullable=False),
        Column("status", String(20), nullable=False),
    )
    meta.create_all(engine)

    # 插入测试数据
    with engine.connect() as conn:
        with conn.begin():
            for i, (title, price, cat) in enumerate([
                ("商品A", 100, "电子"),
                ("商品B", 200, "电子"),
                ("商品C", 50, "食品"),
                ("商品D", 300, "服装"),
            ], start=1):
                conn.execute(
                    text("INSERT INTO products VALUES (:id, :title, :price, :cat, 'online')"),
                    {"id": i, "title": title, "price": price, "cat": cat},
                )
    return products

def test_where_multiple_conditions(engine, products_table):
    """验证多个 where 条件正确叠加为 AND"""
    p = products_table
    stmt = (
        select(p).where(p.c.category == "电子").where(p.c.unit_price >= 150)
    )
    with engine.connect() as conn:
        results = conn.execute(stmt).all()
        assert len(results) == 1
        assert results[0].title == "商品B"

def test_count_group_by(engine, products_table):
    """验证聚合 + 分组"""
    p = products_table
    stmt = (
        select(p.c.category, func.count().label("cnt"))
        .group_by(p.c.category)
        .order_by(func.count().desc())
    )
    with engine.connect() as conn:
        results = conn.execute(stmt).all()
        assert len(results) == 3
        assert results[0].category == "电子"
        assert results[0].cnt == 2

def test_sql_injection_safe(engine, products_table):
    """验证参数化查询的安全性（注入字符串不会生效）"""
    p = products_table
    malicious = "'; DROP TABLE products; --"
    stmt = select(p).where(p.c.category == malicious)
    compiled = stmt.compile()
    # 确保参数值在 params 中，不在 SQL 文本中
    assert "DROP TABLE" not in str(compiled)
    assert compiled.params.get("category_1") == malicious

def test_select_column_subset(engine, products_table):
    """验证只选择部分列"""
    p = products_table
    stmt = select(p.c.id, p.c.title).where(p.c.category == "食品")
    with engine.connect() as conn:
        row = conn.execute(stmt).fetchone()
        assert row.id == 3
        assert row.title == "商品C"

四、项目总结

优点与缺点

对比维度	字符串拼接 SQL	SQL Expression Language
SQL 注入防护	人工保证，易遗漏	参数化查询自动绑定，不可能注入
条件化组装	手动管理 %s 顺序和类型	Python 对象链式拼接，类型自动管理
可读性	SQL 字符串长且混杂 Python 逻辑	结构化的 Python 代码，IDE 可导航
复用性	剪贴复制 SQL 片段	函数抽取，stmt 对象可以传递和组合
调试	需要抓包或开 general log	compile() 直接输出 SQL 和参数
学习成本	低（会 SQL 就行）	中（需适应 Python 对象式写 SQL）

适用场景

1. 带多重可选条件筛选的后台列表查询（最常见）。
2. 报表类查询——聚合、分组、子查询的组合封装。
3. 需要在 Core 层执行但不想写裸 SQL 的场景（批量插入、复杂更新等）。
4. 需要跨数据库兼容的查询（Dialect 层自动翻译）。
5. 需要在 ORM 之外做高性能只读查询的场景（Core 比 ORM 轻量）。

不推荐使用的场景：

1. 简单固定 SQL——text("SELECT * FROM table WHERE id=1") 更直接。
2. 数据库厂商特有的 SQL 特性（如 PostgreSQL 的 WINDOW 函数、递归 CTE），这些可能需要 text() 配合使用。

注意事项

1. select() 的使用有 select() 和 select("*") 的区别：select("*") 会报错。要么传 table 对象（所有列），要么传具体列名。
2. Core 层的 Row 对象是命名元组：支持 row.id 和 row[0] 两种访问方式，也可以用 row._mapping 转为字典。
3. 条件中注意 None 值：column == None 不会生成 SQL IS NULL，而是生成 = NULL（永远不匹配）。正确写法是 column.is_(None)。
4. 不要在生产代码中用 compile(literal_binds=True)：这是为调试设计的，不仅有性能开销，而且对于非字面量的参数可能会报错。

常见踩坑经验

案例 1：func.now() 被缓存为静态值

• 现象：批量插入数据时，所有行的 created_at 都是同一个时间戳。
• 根因：func.now() 在 Python 进程启动时求值了一次，后续使用缓存的结果。
• 修复：使用 sqlfunc.now() 或者 text("NOW()") 作为 server_default。

案例 2：limit() 与 offset() 在子查询中的顺序

• 现象：子查询的 LIMIT 行为与预期不符。
• 根因：子查询中的 LIMIT 需要配合 ORDER BY 使用，否则结果不保证稳定。
• 修复：子查询内始终加 ORDER BY 再 LIMIT，或者使用窗口函数。

案例 3：where 条件的组合使用了错误的逻辑运算符

• 现象：写了三个 .where() 期望是 (A OR B) AND C，实际上生成了 A AND B AND C。
• 根因：.where() 永远追加 AND。
• 修复：or_() 需要显式使用：select(...).where(or_(A, B)).where(C)。

思考题

1. 假设商品表有 100 万行，你需要实现一个"分类 + 关键词"搜索接口。前端传入 keyword 和 category 两个可选参数。请问：如果用户只传 keyword（如 “蓝牙”），能否使用 ILIKE '%蓝牙%'？这种查询在大数据量下的潜在性能问题和解决方案是什么？（提示：PostgreSQL 全文搜索、trigram 索引）

2. 代码中有一个函数返回 select(...) 对象，调用方在得到这个对象后又追加了 .limit(20)。但如果 select() 对象本身已经包含 .limit(10)，最终是 10 还是 20？limit() 的行为是覆盖还是追加？请通过代码验证并说明。

参考答案参见附录 E。

延伸阅读与资源

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