适用场景

导入程序分别拿到商品编号和库存数量，然后逐项配对更新。日志显示任务正常结束，但输入有三个编号，数据库只更新了两个。下面用内存字典模拟库存写入，讨论 Python 3.10 及以上版本中的配对校验；示例不需要第三方依赖。

这里有两个独立问题：默认 zip 会在最短输入耗尽时结束；换成 strict=True 后，长度异常可能在若干条记录已经写入之后才出现。修复必须同时覆盖“发现不一致”和“校验失败前没有写入”。

一、先复现静默丢行

product_ids = [101, 102, 103]
quantities = [8, 9]
stock = {}
for product_id, quantity in zip(product_ids, quantities):
    stock[product_id] = quantity
print(stock)
assert stock == {101: 8, 102: 9}

103 没有被处理，也没有异常。排查时先核对两侧数据各自经过的过滤、分页和错误处理：一侧跳过空值、另一侧保留原始编号，是常见来源。不要只记录循环执行次数，那只能证明配对后的条数。

如果两侧来自不同 SQL 查询，即使长度相同，也不能证明顺序一致。应由同一条记录携带编号和数量，或者通过稳定唯一键关联；单纯补一个长度检查无法修复错位更新。

二、strict=True 为什么仍可能留下部分写入

stock = {}
try:
    for product_id, quantity in zip([101, 102, 103], [8, 9], strict=True):
        stock[product_id] = quantity
except ValueError:
    print("输入长度不一致，当前写入条数：", len(stock))
assert stock == {101: 8, 102: 9}

zip 返回惰性迭代器。前两对元素合法，循环体已经执行；继续取第三对时才发现长度不同。构造 zip(..., strict=True) 本身不会完成校验，提前 break 也可能让尾部异常永远没有机会触发。

Python 官方说明了默认最短截断、惰性求值和严格模式的行为，strict 从 3.10 起可用：zip 官方文档。严格模式负责暴露长度差异，不提供事务回滚。

三、有界批次先完整校验，再交给写入边界

本例约定：商品编号为正整数，库存为非负整数；拒绝布尔值、重复编号；最多 1000 对。两侧都为空时视为合法空批次。实际服务还应按业务设置数值上限、请求体大小和权限校验。

保存下面代码为 stock_batch.py。prepare_updates 只构建计划，不接触数据库。长度、值或来源异常都发生在返回计划之前。

from collections.abc import Iterable

MAX_BATCH_SIZE = 1000


def prepare_updates(
    product_ids: Iterable[int], quantities: Iterable[int]
) -> tuple[tuple[int, int], ...]:
    """在有界批次内完成配对和值校验，避免边校验边写入。"""
    updates: list[tuple[int, int]] = []
    seen: set[int] = set()
    for product_id, quantity in zip(product_ids, quantities, strict=True):
        if len(updates) >= MAX_BATCH_SIZE:
            raise ValueError("批次超过条数上限")
        if type(product_id) is not int or product_id <= 0:
            raise ValueError("商品编号必须为正整数")
        if type(quantity) is not int or quantity < 0:
            raise ValueError("库存必须为非负整数")
        if product_id in seen:
            raise ValueError("批次存在重复商品编号")
        seen.add(product_id)
        updates.append((product_id, quantity))
    return tuple(updates)


def update_memory_stock(
    stock: dict[int, int],
    product_ids: Iterable[int],
    quantities: Iterable[int],
) -> None:
    """用内存替身展示：准备阶段失败时不会进入写入阶段。"""
    updates = prepare_updates(product_ids, quantities)
    stock.update(updates)

返回元组将本次计划固定下来，后续不再读取原始迭代器。1000 条限制防止普通连续输入被无限收集，但不能限制一个 next() 调用的阻塞时间，也不能约束任意来源的内部内存占用；来源自身必须有超时和资源限制。检测长度差异时迭代器可能已经前移，失败后不能直接复用同一迭代器重试。

正式数据库写入应在计划成功后开启事务，用参数化批量语句提交；中途约束失败必须整体回滚。上面的字典替身只证明准备失败不写入，不证明数据库事务、并发竞争或外部 API 的原子性。跨服务更新需要额外设计幂等键和失败恢复，不能靠本地列表解决。

四、回归测试必须断言原库存没有变化

保存为 test_stock_batch.py，与上一文件放在同一目录：

import unittest

from stock_batch import prepare_updates, update_memory_stock


class StockBatchTests(unittest.TestCase):
    """覆盖完整批次、输入边界和准备失败时的写入隔离。"""

    def test_valid_generators(self) -> None:
        self.assertEqual(
            prepare_updates(iter([101, 102]), iter([0, 9])),
            ((101, 0), (102, 9)),
        )

    def test_empty_batch(self) -> None:
        self.assertEqual(prepare_updates([], []), ())

    def test_length_mismatch_preserves_stock(self) -> None:
        for ids, values in [([101, 102], [8]), ([101], [8, 9]), ([], [8])]:
            with self.subTest(ids=ids, values=values):
                stock = {101: 3}
                with self.assertRaises(ValueError):
                    update_memory_stock(stock, ids, values)
                self.assertEqual(stock, {101: 3})

    def test_invalid_values(self) -> None:
        for ids, values in [([0], [8]), ([True], [8]), ([101], [-1]),
                            ([101], [True]), ([101], ["8"])]:
            with self.subTest(ids=ids, values=values):
                with self.assertRaises(ValueError):
                    prepare_updates(ids, values)

    def test_duplicate_id(self) -> None:
        with self.assertRaises(ValueError):
            prepare_updates([101, 101], [8, 9])

    def test_batch_limit(self) -> None:
        self.assertEqual(len(prepare_updates(range(1, 1001), [0] * 1000)), 1000)
        with self.assertRaises(ValueError):
            prepare_updates(range(1, 1002), [0] * 1001)

    def test_source_failure_preserves_stock(self) -> None:
        def broken_source():
            """模拟来源读完第一条后失败。"""
            yield 101
            raise OSError("模拟输入读取失败")

        stock = {101: 3}
        with self.assertRaises(OSError):
            update_memory_stock(stock, broken_source(), [8, 9])
        self.assertEqual(stock, {101: 3})


if __name__ == "__main__":
    unittest.main()

执行 python -m unittest -v test_stock_batch，应运行 7 个测试方法。不能只断言“抛出了 ValueError”：原始的边写边严格校验实现也会抛错，却已经更新了一部分库存。

本次在 Python 3.14.6 上执行上述 7 项测试全部通过；将准备函数中的严格模式改回默认截断行为后，3 个长度不一致子用例失败，恢复后再次全通过。两个故障复现片段也已执行。UTF-8 无 BOM、代码语法和代码围栏检查通过；环境未安装 Ruff，未执行其格式与静态检查。

五、上线前检查与实践边界

检查所有并行数组的构造路径，包括失败项是否被单独过滤；优先把编号和数据保存在同一个记录中。确实要求一一对应时启用严格校验，不要用 zip_longest 随意填补库存。若业务允许忽略多余数据，应显式记录该规则和丢弃数量。

拒绝批次时记录 batch_id、error_type、validated_count 等结构化字段，固定消息使用中文，避免把原始商品清单或敏感请求体写入日志。失败不得被包装成正常导入完成。

本文本地验证使用 Python 标准库与内存替身，未连接生产数据库、真实文件流或远端接口，也未验证数据库提交失败、并发库存更新和生产权限。先完整准备计划可以隔离输入失败；实际写入的完整性仍由事务和业务并发控制保证。

总结

默认 zip 会让较长输入的尾部消失；strict=True 会发现长度不同，但发现时间可能晚于写入。对规模可控的批次，完整消费并校验所有配对后再进入写入阶段，配合失败时原状态不变的测试，才能防住这两种故障。