Python订单数据库教程：JOIN聚合索引与事务回滚实现

脚本专家

在Python开发中，订单系统是练习数据库设计、SQL查询和事务控制的经典场景。本文基于SQLite环境，从零搭建用户、商品、订单、订单项四张表，重点讲解关联查询、聚合统计、索引优化和事务回滚，适合刚学完Python基础并希望掌握数据库编程的读者。

一、多对多关联与中间表设计
在进入订单系统前，先回顾上一篇的多对多关系。图书与作者是多对多关系，必须通过中间表book_authors维护。表中使用联合主键(book_id, author_id)防止重复关联，外键保证引用的编号真实存在，并设置ON DELETE CASCADE实现级联删除。

建表SQL如下：

2. CREATE TABLE authors (
3.     id INTEGER PRIMARY KEY AUTOINCREMENT,
4.     name TEXT NOT NULL UNIQUE
5. );
7. CREATE TABLE book_authors (
8.     book_id INTEGER NOT NULL,
9.     author_id INTEGER NOT NULL,
10.     PRIMARY KEY (book_id, author_id),
11.     FOREIGN KEY (book_id) REFERENCES books(id) ON DELETE CASCADE,
12.     FOREIGN KEY (author_id) REFERENCES authors(id) ON DELETE CASCADE
13. );

复制代码

参数化关联函数：

2. def add_author_to_book(book_id, author_id):
3.     with connect() as connection:
4.         connection.execute(
5.             "INSERT INTO book_authors (book_id, author_id) VALUES (?, ?)",
6.             (book_id, author_id),
7.         )

复制代码

验证时注意：SQLite默认可能关闭外键检查，必须执行PRAGMA foreign_keys = ON；否则删除book后关联记录可能残留。

二、订单项必须是独立实体
在设计订单表时，order_items是独立表，而不是在orders中存一个商品列表。原因在于：每个订单可能包含多个商品，且每个商品在成交时的价格必须快照到订单项中。如果下单后商品涨价，历史订单仍应显示当时的成交价，不能去读当前价格。

订单项建表语句：

2. CREATE TABLE order_items (
3.     id INTEGER PRIMARY KEY,
4.     order_id INTEGER NOT NULL,
5.     product_id INTEGER NOT NULL,
6.     quantity INTEGER NOT NULL CHECK (quantity > 0),
7.     price NUMERIC NOT NULL CHECK (price > 0),
8.     FOREIGN KEY (order_id) REFERENCES orders(id) ON DELETE CASCADE,
9.     FOREIGN KEY (product_id) REFERENCES products(id)
10. );

复制代码

price字段保存成交时的单价，quantity必须大于0，这两项都属于业务约束，应放到数据库层面。

三、JOIN关联查询与聚合统计
订单列表需要同时显示用户姓名、商品名称、下单数量、单价和小计金额。这需要四张表JOIN：

2. SELECT o.id AS order_id, u.name, p.name AS product_name,
3.        oi.quantity, oi.price, oi.quantity * oi.price AS line_total
4. FROM orders o
5. JOIN users u ON u.id = o.user_id
6. JOIN order_items oi ON oi.order_id = o.id
7. JOIN products p ON p.id = oi.product_id
8. WHERE o.id = ?;

复制代码

INNER JOIN只返回匹配行。如果某个用户没有订单，JOIN后这个用户不会出现；需要统计所有用户消费情况时，必须使用LEFT JOIN保留左表所有行：

2. SELECT u.id, u.name, COUNT(o.id) AS order_count,
3.        COALESCE(SUM(o.total_amount), 0) AS spent
4. FROM users u
5. LEFT JOIN orders o ON o.user_id = u.id
6. GROUP BY u.id, u.name
7. ORDER BY spent DESC;

复制代码

COALESCE将NULL转为0，避免没有订单的用户显示NULL。COUNT只统计非NULL的o.id，所以无订单用户order_count为0。

四、索引：提升查询性能的代价
订单表经常按用户ID和创建时间排序查询，可以创建复合索引：

2. CREATE INDEX idx_orders_user_created
3. ON orders (user_id, created_at DESC);

复制代码

索引加快读取，但每次INSERT、UPDATE、DELETE都需要额外维护索引，占用存储空间。不要提前建索引，应在真实查询场景中观察慢SQL后再创建。

五、事务：下单必须全部成功或全部回滚
下单流程包含三个写操作：插入订单、插入订单项、扣减库存。任何一步失败都不允许留下半条业务数据，因此必须使用事务。下面代码使用SQLite的BEGIN/COMMIT/ROLLBACK实现：

2. from datetime import datetime, timezone
4. def create_order(connection, user_id, items):
5.     try:
6.         connection.execute("BEGIN")
7.         cursor = connection.execute(
8.             "INSERT INTO orders (user_id,total_amount,created_at) VALUES (?,0,?)",
9.             (user_id, datetime.now(timezone.utc).isoformat()),
10.         )
11.         order_id = cursor.lastrowid
12.         total = 0
13.         for product_id, quantity in items:
14.             product = connection.execute(
15.                 "SELECT name,price,stock FROM products WHERE id = ?",
16.                 (product_id,),
17.             ).fetchone()
18.             if product is None:
19.                 raise ValueError("商品不存在")
20.             if product["stock"] < quantity:
21.                 raise ValueError(f"{product['name']}库存不足")
22.             connection.execute(
23.                 "INSERT INTO order_items(order_id,product_id,quantity,price) VALUES (?,?,?,?)",
24.                 (order_id, product_id, quantity, product["price"]),
25.             )
26.             connection.execute(
27.                 "UPDATE products SET stock=stock-? WHERE id=?",
28.                 (quantity, product_id),
29.             )
30.             total += product["price"] * quantity
31.         connection.execute(
32.             "UPDATE orders SET total_amount=? WHERE id=?",
33.             (total, order_id),
34.         )
35.         connection.commit()
36.         return order_id
37.     except Exception:
38.         connection.rollback()
39.         raise

复制代码

注意两点：一是业务异常先raise，让调用者感知失败原因，不能回滚后假装成功；二是connection.execute返回的cursor.lastrowid用于获取新插入订单的ID。

验证回滚时，准备一个库存充足和一个库存不足的商品同时下单。预期结果是：orders和order_items没有新增记录，库存充足的商品库存也没有减少。不要只检查是否抛出异常，还要查询三张表确认没有“半条”数据残留。

六、UPDATE和DELETE必须检查影响行数
在Python中执行UPDATE或DELETE后，cursor.rowcount表示受影响行数。若为0，通常说明目标记录不存在，不能直接认为成功：

2. cursor = db.execute('UPDATE tasks SET done = ? WHERE id = ?', (1, task_id))
3. if cursor.rowcount == 0:
4.     raise NotFound('任务不存在')
5. db.commit()

复制代码

课后练习题：实现cancel_order(connection, order_id)，要求只能取消未取消订单；把订单项中的数量加回库存，再修改订单状态；任意一步失败则全部回滚。也可以把“完成任务”和“写入审计日志”放进同一事务，并编写成功、不存在、审计失败三组测试。下一篇将会把SQLite环境迁移到MySQL，对比不同数据库的事务和连接配置差异。

七、完整模块与验收总结
本文涉及的完整模块文件应替换旧文件后重新运行测试。新增函数的事务边界、错误处理是阅读重点。最终应掌握：区分一对多与多对多；订单项保存价格快照；编写JOIN、GROUP BY和聚合查询；理解索引读写权衡；事务成功路径提交、失败路径回滚。这些技能是构建可靠Python数据应用的基石。

热心网友1

感谢分享！内容很扎实，特别是“订单项必须独立实体”那段，用价格快照来解释很到位，比很多教程说“规范化”更让人理解。另外复合索引和LEFT JOIN聚合的示例也很实用。 有个小问题：最后的事务部分，`create_order` 函数写到 `for product_id, quantity in` 就截断了，是不是漏了后半段？很想看完整的回滚逻辑是怎么处理的，比如库存不足时怎么抛异常回滚。希望楼主能补充一下，谢谢！

热心网友1

这个教程写得很扎实，尤其是把订单项作为独立表、价格快照这点，确实是在实际项目里容易踩坑的地方。用 SQLite 做示例也很合适，初学者能直接跑起来看效果。事务部分用 BEGIN/COMMIT/ROLLBACK 手动控制，比默认的隐式提交更清楚，能让人真正理解“要么全成功，要么全回滚”的含义。唯一想补充的是，在实际生产环境里 SQLite 的并发写可能是个问题，不过作为学习数据库思想的教程完全够用。建议可以再加一个常见问题小节，比如外键没生效或者死锁怎么排查，会更完整。

热心网友1

教程很实用，正好最近在练数据库这块。之前自己写订单表的时候确实没想过价格要快照到订单项里，都是下单后直接去查商品当前价格，看了一下确实会有历史订单金额对不上的问题。 还有LEFT JOIN和COALESCE那部分讲得很清楚，之前统计用户消费时老是被空值困扰，原来问题出在这里。 想问一下，那个BEGIN显式开启事务的方式，和Python sqlite3默认的隐式事务行为有区别吗？我看不少人直接依赖上下文管理器自动提交，遇到异常再rollback，不知道哪种方式更稳妥？