场景与现象
日报接口按店铺汇总实收金额。正常营业日返回整数分，某天没有订单时却返回 null，下游执行金额加法失败。开发把 SQL 改成 SUM(COALESCE(amount_cents, 0))，有订单但金额全空的场景恢复了，没有订单的场景依然失败。
本文用 SQLite 和 Python 标准库复现，不需要线上数据库。金额统一按整数分保存；接口约定“没有订单”为零，“存在订单但金额缺失”为数据质量错误。后一个约定很关键：不能为了消除 null，把未知金额直接当成没有收入。
先区分三种输入
不要先改前端格式化。对同一筛选条件同时查看行数、有效金额数和原始合计：
SELECT COUNT(*) AS row_count,
COUNT(amount_cents) AS valued_count,
SUM(amount_cents) AS raw_total
FROM payments
WHERE shop_id = :shop_id;
row_count=0 表示没有匹配记录；row_count>valued_count 表示记录中存在金额缺失；两者相等而合计为零，才可能是零金额或退款抵消。定位时保留原来的店铺条件、支付状态和时间范围，否则统计口径不同会掩盖问题。
SQLite 的 SUM 忽略 NULL，无有效输入时返回 NULL；TOTAL 返回浮点数，包括空输入时的 0.0。金额报表因此不能随手改用 TOTAL，否则返回类型和精度边界也会变化。依据：SQLite 聚合函数文档。
为什么行内补零没有修好
下面三条表达式处理的是不同阶段：
SELECT SUM(amount_cents),
SUM(COALESCE(amount_cents, 0)),
COALESCE(SUM(amount_cents), 0)
FROM payments
WHERE shop_id = :shop_id;
行内 COALESCE 只能替换实际存在的输入行中的 NULL。筛选后一行都没有时，它没有处理对象。外层 COALESCE 则处理聚合结果，可以把无有效输入的结果转换为零。
但外层补零也会把“全为空值”变成零。修复应同时检查两个计数，再决定能否使用合计。不要在 Python 里用 result or 0 掩盖缺失值，也不要先调用 int(result)：前者丢失业务状态，后者会在 None 上报错。
可直接运行的实现与回归测试
将以下完整代码保存为 test_payment_summary.py，执行 python -m unittest -v test_payment_summary。数据库连接由调用方提供，测试只用内存库，查询值通过参数绑定传入。
"""验证订单金额汇总的空集与缺失值边界。"""
import sqlite3
import unittest
from contextlib import closing
SUMMARY_SQL = """
SELECT COUNT(*), COUNT(amount_cents), COALESCE(SUM(amount_cents), 0)
FROM payments
WHERE shop_id = ?
"""
def read_total(connection: sqlite3.Connection, shop_id: int) -> int:
"""无订单返回零，金额缺失则拒绝输出不完整的合计。"""
if type(shop_id) is not int or shop_id <= 0:
raise ValueError("店铺编号必须为正整数")
row_count, valued_count, total = connection.execute(
SUMMARY_SQL, (shop_id,)
).fetchone()
if row_count != valued_count:
raise ValueError("订单金额缺失，无法生成完整合计")
if type(total) is not int:
raise TypeError("汇总金额必须为整数分")
return total
class PaymentSummaryTests(unittest.TestCase):
"""覆盖无记录、缺失、退款、类型及分组边界。"""
def setUp(self) -> None:
self.connection = sqlite3.connect(":memory:")
self.addCleanup(self.connection.close)
self.connection.execute("""
CREATE TABLE payments (
shop_id INTEGER NOT NULL,
amount_cents INTEGER
CHECK (amount_cents IS NULL
OR typeof(amount_cents) = 'integer')
)
""")
def add_amounts(self, amounts: list[int | None]) -> None:
"""保留可空金额以模拟历史脏数据。"""
self.connection.executemany(
"INSERT INTO payments VALUES (?, ?)",
[(1, amount) for amount in amounts],
)
def test_empty(self) -> None:
self.assertEqual(read_total(self.connection, 1), 0)
def test_integer_total(self) -> None:
self.add_amounts([1200, 35])
self.assertEqual(read_total(self.connection, 1), 1235)
def test_refund_cancellation(self) -> None:
self.add_amounts([1200, -1200, 0])
self.assertEqual(read_total(self.connection, 1), 0)
def test_missing_amount(self) -> None:
for amounts in ([None], [1200, None]):
with self.subTest(amounts=amounts):
self.connection.execute("DELETE FROM payments")
self.add_amounts(amounts)
with self.assertRaisesRegex(ValueError, "金额缺失"):
read_total(self.connection, 1)
def test_shop_isolation(self) -> None:
self.add_amounts([1200])
self.assertEqual(read_total(self.connection, 2), 0)
def test_invalid_shop(self) -> None:
for shop_id in (0, -1, True, "1 OR 1=1"):
with self.subTest(shop_id=shop_id):
with self.assertRaises(ValueError):
read_total(self.connection, shop_id)
def test_storage_constraint(self) -> None:
with self.assertRaises(sqlite3.IntegrityError):
self.connection.execute("INSERT INTO payments VALUES (1, 1.5)")
def test_integer_overflow_is_not_zero(self) -> None:
self.add_amounts([9223372036854775807, 1])
with self.assertRaises(sqlite3.OperationalError):
read_total(self.connection, 1)
def test_grouped_empty_has_no_row(self) -> None:
rows = self.connection.execute("""
SELECT shop_id, COALESCE(SUM(amount_cents), 0)
FROM payments GROUP BY shop_id
""").fetchall()
self.assertEqual(rows, [])
if __name__ == "__main__":
with closing(sqlite3.connect(":memory:")) as connection:
connection.execute("CREATE TABLE payments(amount_cents INTEGER)")
print("空表三种写法：", connection.execute("""
SELECT SUM(amount_cents), SUM(COALESCE(amount_cents, 0)),
COALESCE(SUM(amount_cents), 0) FROM payments
""").fetchone())
unittest.main()
直接执行文件时，复现行应为 (None, None, 0)。随后应执行九项测试，而不是只运行查询演示。test_empty 是原始故障的回归用例；test_missing_amount 防止补零吞掉数据异常；溢出测试要求错误继续传播，不能把计算失败当作营业额零。
表中的 CHECK 约束检查的是 SQLite 实际存储类型，不能替代接口入参校验。示例允许 NULL 是为了复现旧数据；生产若业务不允许缺失，应先完成历史数据核对，再考虑 NOT NULL 迁移。
分组查询还有另一层边界
上面的汇总没有 GROUP BY，也没有 HAVING。若按店铺分组，空输入根本不会产生店铺组，外层 COALESCE 也无法凭空生成店铺行。带 HAVING 的查询还可能过滤掉聚合结果。相关规则见 SQLite SELECT 文档。
需要每天展示全部店铺时，应由店铺清单或日历维表确定输出集合，先汇总支付表，再关联汇总结果。关联后的“没有汇总行”和“汇总行内金额缺失”仍要分别处理。不要通过往支付表插入零元假订单补齐报表，那会污染订单数量、审计和对账口径。
上线步骤与验证边界
- 在只读查询中增加总记录数和有效金额数，确认失败日期确实为空集，还是订单金额缺失。
- 明确接口零值与缺失值的合同；仅在业务认可的空集场景返回零。
- 保持原有状态、时间区间与权限筛选，将补零放到聚合外层，加入缺失金额检查。
- 比较新旧查询在正常日、无订单日、退款抵消日和脏数据日的输出。只读汇总改动不涉及数据回填；回滚时恢复旧查询即可。
本文只验证内存 SQLite 的计算语义与 Python 返回值，没有验证生产 ORM、并发读取、索引计划、真实支付状态或跨库兼容性。整数 SUM 超出 SQLite 范围会失败，外层 COALESCE 不能修复溢出；大额账务应另行设计数值范围与精确计算方案。
总结
排查金额 null 时，先查输入行数和有效值数。行内补零、聚合后补零以及补齐缺失分组分别解决不同问题。保留这些状态的区别，才能让无订单返回零，同时让金额缺失和计算失败被及时发现。
验证记录：在 Python 3.14.6、SQLite 3.50.4 上，正文九项 unittest 全部通过。将外层补零换回行内补零后，两个空集用例触发类型校验错误，其余七项通过，确认回归测试能检出原始缺陷。正文 UTF-8、代码语法和围栏检查通过；环境未安装 Ruff，未执行 Ruff 检查，也未新增依赖。
Discussion
评论