一个报表接口连续执行两次：第一次查询订单，第二次查询库存。库存请求却带上了订单请求追加的字段。重启进程后暂时恢复，压测时又复现。数据库查询本身没有问题，污染发生在构造查询参数之前。
这类故障不一定需要并发：同一进程中两个顺序调用就能触发。本文用纯 Python 标准库复现并修复，适用于长期运行的 Web 服务、任务 Worker 和复用配置模板的脚本。示例使用 Python 3.10 及以上版本。
先把故障缩到两个顺序调用
下面是故障复现代码，不能直接用于业务。保存为 reproduce.py 后执行 python reproduce.py：
def build_columns(resource: str, columns: list[str] = []) -> list[str]:
"""故障示例：每次调用意外复用了同一个默认列表。"""
columns.append(resource)
return columns
first = build_columns("orders")
second = build_columns("inventory")
print(first is second)
print(second)
template = {"columns": ["id"], "options": {"limit": 100}}
request_parameters = template.copy()
request_parameters["columns"].append("amount")
print(template["columns"])
print(request_parameters["columns"] is template["columns"])
输出如下：
True
['orders', 'inventory']
['id', 'amount']
True
第一处问题来自函数默认值。函数定义执行时创建默认列表，以后省略 columns 的调用会复用它。变量 first 也引用该列表，所以第二次调用还能改变第一次调用的结果。
第二处问题来自浅拷贝。template.copy() 创建了外层字典，但其中的列表和嵌套字典仍然共享。检查 id(request_parameters) 与 id(template) 不同，不能证明内部字段已经隔离；这里应检查 request_parameters["columns"] is template["columns"]。
在服务中怎么定位
先固定到同一个 Worker，按 A 请求、B 请求、A 请求的顺序调用，比较最终发给数据库或下游服务的参数结构。若串数据只发生在同一进程，优先检查对象是否跨请求复用，而不是先调大连接池。
沿参数构造链寻找这些位置：函数签名中的 =[]、={}，类属性上的可变容器，模块级模板，缓存返回的可变对象，以及 dict.copy()、list.copy() 后继续修改嵌套字段的代码。
可在受控诊断中记录 worker_id、request_id、template_version 和 shares_columns。固定日志消息使用“请求参数引用检查”。shares_columns 来自对象身份比较，不要把完整筛选条件、客户标识或请求体写入日志。id() 只适合当前进程的短时诊断，不能作为跨进程标识。
还要区分“两个请求复用了对象”和“调用方传入对象后被函数修改”。即使把默认参数改成 None，下面这种写法仍然会改动调用方列表：先把传入列表赋给局部变量，再执行 append()。必须明确函数是否有权修改输入。
修复：按字段构造独立的参数对象
这里约定参数只有两个字段：columns 是最多 20 个非空字符串，limit 是 1 到 1000 的整数。省略 columns 时使用 id，显式传入空列表表示错误；字符串、布尔值等意外类型必须拒绝。
保存以下完整代码为 query_parameters.py。入口接收已经解析出的 Python 对象，校验后再构造内部模型。函数每次返回独立列表，不修改调用方数据。
from dataclasses import dataclass, field
@dataclass
class QueryParameters:
"""保存单次查询的内部参数，不充当未校验的请求模型。"""
columns: list[str] = field(default_factory=lambda: ["id"])
limit: int = 100
def build_parameters(
columns: object = None,
limit: object = 100,
) -> QueryParameters:
"""校验外部字段并复制容器，避免请求之间共享可变状态。"""
if type(limit) is not int or not 1 <= limit <= 1000:
raise ValueError("limit 必须是 1 到 1000 的整数")
if columns is None:
return QueryParameters(limit=limit)
if not isinstance(columns, list) or not 1 <= len(columns) <= 20:
raise ValueError("columns 必须是包含 1 到 20 项的列表")
copied_columns: list[str] = []
for column in columns:
if (
not isinstance(column, str)
or not column.strip()
or len(column) > 64
):
raise ValueError("列名必须是长度不超过 64 的非空字符串")
copied_columns.append(column)
return QueryParameters(columns=copied_columns, limit=limit)
default_factory 在需要默认字段值时调用工厂，因此两次 QueryParameters() 得到不同的列表。注意，default_factory=lambda: shared_list 仍然返回共享对象，不会自动复制。Python 官方文档给出了默认工厂的调用约定。
copied_columns 只复制一层就够，因为校验后元素全部是字符串。若字段变成 list[dict]，必须继续逐字段构造内部对象，不能照搬这段复制逻辑。type(limit) is int 是刻意的：Python 的 bool 是 int 的子类，但接口不应把 True 当成限制条数 1。
模型保持可变，方便单次请求继续追加经过授权的列。如果对象要放进全局缓存，建议改用 tuple[str, ...] 等不可变结构。仅给包含列表的 dataclass 加上 frozen=True，并不能阻止列表执行 append()；它限制的是字段赋值。dataclasses 官方说明描述了这一边界。
上述校验只建立对象所有权和基础格式边界。如果 columns 最终决定 SQL 列名，必须再映射到服务端允许的列集合，并执行字段权限校验；数据值使用数据库参数绑定，不能将客户端列名直接拼接进 SQL。
用回归测试锁住三种隔离关系
保存为 test_query_parameters.py，与实现放在同一目录，然后执行 python -m unittest -v test_query_parameters：
import unittest
from query_parameters import QueryParameters, build_parameters
class QueryParametersTests(unittest.TestCase):
"""验证实例、调用方输入以及不同请求之间的对象隔离。"""
def test_default_instances_are_independent(self) -> None:
first = QueryParameters()
second = QueryParameters()
first.columns.append("amount")
self.assertEqual(second.columns, ["id"])
self.assertIsNot(first.columns, second.columns)
def test_default_requests_are_independent(self) -> None:
first = build_parameters()
second = build_parameters()
first.columns.append("amount")
self.assertEqual(second.columns, ["id"])
def test_input_changes_do_not_change_result(self) -> None:
source = ["id"]
result = build_parameters(source)
source.append("amount")
self.assertEqual(result.columns, ["id"])
def test_result_changes_do_not_change_input(self) -> None:
source = ["id"]
result = build_parameters(source)
result.columns.append("amount")
self.assertEqual(source, ["id"])
def test_same_input_can_build_independent_requests(self) -> None:
source = ["id"]
first = build_parameters(source)
second = build_parameters(source)
first.columns.append("amount")
self.assertEqual(second.columns, ["id"])
def test_limit_boundaries(self) -> None:
for limit in (1, 1000):
with self.subTest(limit=limit):
self.assertEqual(build_parameters(limit=limit).limit, limit)
def test_invalid_limits(self) -> None:
for limit in (0, 1001, True, 1.5, "100", None):
with self.subTest(limit=limit):
with self.assertRaises(ValueError):
build_parameters(limit=limit)
def test_invalid_columns(self) -> None:
for columns in ([], "id", {}, [1], [" "], ["x" * 65], ["id"] * 21):
with self.subTest(columns=columns):
with self.assertRaises(ValueError):
build_parameters(columns)
def test_column_boundaries(self) -> None:
self.assertEqual(build_parameters(["x" * 64]).columns, ["x" * 64])
self.assertEqual(len(build_parameters(["id"] * 20).columns), 20)
if __name__ == "__main__":
unittest.main()
不能只断言两个返回值相等。两个列表内容相同，可能仍然引用同一对象。上面的测试实际修改其中一方，再检查另一方保持原样，覆盖了“默认实例之间”“调用方与返回值之间”“复用同一输入的请求之间”三种关系。
这九项测试验证纯 Python 参数构造边界，不证明框架依赖注入、全局缓存、数据库驱动或多进程服务不存在其他共享状态。接入服务后还应在同一 Worker 上重复执行 A/B/A 请求，并验证数据库收到的实际参数符合预期。
为什么不直接到处加 deepcopy
deepcopy() 对普通嵌套容器通常能递归复制，但它不替代字段校验，也不等于复制一份独立的数据库连接或文件句柄。自定义对象还可能定义复制行为，复制整个业务上下文也可能扩大 CPU 和内存成本。copy 官方文档说明了浅拷贝、深拷贝和不可复制资源的差异。
对于本文这种字段少、结构确定的请求参数，按字段重新构造更容易审核：哪些数据被接受，哪些字段归当前请求所有，都能直接看出来。确实需要深拷贝大型嵌套模板时，应先确认模板只包含支持复制的数据类型，再针对嵌套字段写修改隔离测试并测量开销。
上线与预防
修复后先重建受污染的内存模板或替换受影响 Worker。只修复新请求的复制逻辑，不会让已经被改坏的模块级模板自动恢复。涉及用户可见结果时，还要根据业务审计确认历史响应是否需要纠正。
代码审查重点看可变对象的所有权：全局模板是否只读、缓存结果是否会被调用方修改、构造函数是否保留外部容器引用。dataclass 的类型标注不会自动进行运行时类型校验，默认工厂也只负责未显式传参的情况；业务入口仍然需要检查输入。
本地验证应同时保留故障复现和修复后的测试。交付前执行已有工程的格式化、静态检查及相关测试；如果环境缺少工具，记录缺项，不能把“语法通过”说成“所有检查通过”。
总结
参数串数据的关键线索是共享引用。函数默认容器会跨调用复用，浅拷贝只分离外层容器，传入的列表也可能被内部代码意外修改。修复要落实到对象所有权：默认字段用独立工厂、外部输入先校验再重建、返回值与调用方容器隔离，并用实际修改一方的测试证明另一方不受影响。
Discussion
评论