别再写嵌套if了：用Python 3.12的match-case重构多源数据清洗管线

一、过去是怎么做的，以及它为什么脆弱

面对来自不同API的异构JSON，通常的思路是：先统一解码成字典，然后根据某个特征字段（比如type或source）来判断是哪个供应商，再针对每个供应商写解析逻辑。比如：

def parse_message(raw: dict):
    if 'orderId' in raw and 'amount' in raw:
        # 可能是供应商A
        return parse_supplier_a(raw)
    elif 'order_id' in raw and 'total' in raw:
        # 可能是供应商B
        return parse_supplier_b(raw)
    elif 'ref' in raw and isinstance(raw.get('items'), list):
        # 供应商C
        return parse_supplier_c(raw)
    else:
        raise ValueError('未知数据格式')

这种写法的主要问题是：判断条件散落在if里，没有明确的“形状”描述。当你需要检查某个字段是否同时满足“存在、是字符串且非空”时，就要在if后面加一串and，可读性快速下降。而且一旦匹配成功，它不会提取出匹配到的具体值，你要在分支里面再次从字典里取值，这又得写好几行。

更隐性的风险在于，这种匹配缺乏穷尽性检查。如果你漏掉了某个分支，或者新来的数据结构碰巧满足了第一个if的条件但实际上并不完全符合预期，解析器可能会悄悄按照错误的分支执行，直到下游报错。

二、match-case 怎么用起来

Python的match-case是从函数式语言借鉴来的特性，但融入了一贯的Python哲学。它可以匹配字面值、变量、序列、映射（字典）甚至自定义类，并且支持在匹配的同时进行析构和绑定。基础语法如下：

def handle(value):
    match value:
        case 200:
            return "成功"
        case int(code) if 400 <= code < 500:
            return f"客户端错误 {code}"
        case str(body) if body.startswith('Error'):
            return f"错误信息: {body}"
        case _:
            return "未知响应"

这里的case后面跟着一个模式，模式里可以出现变量名（如code），如果匹配成功，这些变量就会被赋值，并且可以在if守卫中使用。最后_是通配符，匹配一切。

对于字典（映射模式），我们可以写出非常精确的结构匹配。比如匹配一个包含orderId且amount为非空字符串的字典：

case {'orderId': str(order_id), 'amount': str(amount), **rest} if amount and order_id:
    # 使用 order_id, amount
    ...

这种模式一口气完成了“存在性检查”“类型检查”和“变量绑定”三件事，而且意图一目了然。

三、案例：清洗三个供应商的订单数据

回到我工作的实际场景。我们有三个供应商返回订单列表，每个订单的结构各有不同。供应商A用驼峰命名，字段比较全；供应商B用下划线，而且金额字段叫total，状态用数字表示；供应商C有时候不给金额，而且状态是中文。原来的适配器分散在不同函数里，现在我用match-case全部整合成一个解析函数。

首先定义内部统一的数据类（使用dataclass）：

from dataclasses import dataclass

@dataclass
class UnifiedOrder:
    order_id: str
    amount: float
    status: str  # 只能是 'pending', 'paid', 'cancelled' 之一

然后编写匹配函数。我们假设传入的data是一个字典列表，每个字典代表一个订单，但具体结构未知。匹配时，我们不仅要判断是哪个供应商，还要一次性提取需要的字段并进行转换：

def parse_order(item: dict) -> UnifiedOrder:
    match item:
        # 供应商A：具有 orderId (str), amount (str), status (int)
        case {
            'orderId': str(order_id),
            'amount': str(amount_str),
            'status': int(status_code),
            **_
        } if amount_str and order_id:
            amount = float(amount_str)
            status_map = {1: 'pending', 2: 'paid', 3: 'cancelled'}
            status = status_map.get(status_code, 'pending')
            return UnifiedOrder(order_id=order_id, amount=amount, status=status)

        # 供应商B：具有 order_id (int), total (float), status (str)
        case {
            'order_id': int(order_id),
            'total': float(amount),
            'status': str(status),
            **_
        } if status in ('pending', 'paid', 'cancelled'):
            return UnifiedOrder(order_id=str(order_id), amount=amount, status=status)

        # 供应商C：可能有 'ref' (str), 'value' (float|int), 'state' (str)
        case {
            'ref': str(order_id),
            'value': float(amount),
            'state': str(raw_state),
            **_
        }:
            state_map = {'待支付': 'pending', '已支付': 'paid', '已取消': 'cancelled'}
            status = state_map.get(raw_state, 'pending')
            return UnifiedOrder(order_id=order_id, amount=amount, status=status)

        # 供应商C的变体：没有value字段，金额默认为0
        case {
            'ref': str(order_id),
            'state': str(raw_state),
            **_
        }:
            state_map = {'待支付': 'pending', '已支付': 'paid', '已取消': 'cancelled'}
            status = state_map.get(raw_state, 'pending')
            return UnifiedOrder(order_id=order_id, amount=0.0, status=status)

        case _:
            raise ValueError(f'无法识别的订单格式: {item}')

这一段代码做的事情，如果用传统if-elif和大量isinstance来写，大概需要四十多行，而且每个分支内部还要再做一层字段判断。现在所有结构验证和变量提取都在一行case里完成，新增一个供应商只需要增加一个case块，不会影响其他分支。更重要的，**_通配符明确表示我们不关心其他字段，它们存在与否都无所谓，这比手动pop或者get要直观得多。

四、加入序列匹配处理批量数据

实际API返回的可能是单个对象，也可能是一个列表。我们也可以在match里直接处理序列模式。比如：

def parse_response(payload) -> list[UnifiedOrder]:
    match payload:
        case [*items]:
            # 匹配列表，items 是列表中的所有元素
            return [parse_order(item) for item in items if isinstance(item, dict)]
        case dict():
            # 单个对象
            return [parse_order(payload)]
        case _:
            raise ValueError('响应格式不正确')

序列模式里的[*items]表示匹配任意长度的列表，并把所有元素绑定到items。你也可以匹配特定长度的列表，比如[first, second, *rest]，但在这个场景没必要。这种能力在处理分页数据或批量推送时特别顺手。

五、再进一步：用守卫和自定义类模式做复杂验证

有时候字段之间的约束关系更复杂，比如“如果status是paid，则amount必须大于0”。这种跨字段的业务规则可以写在case的if守卫里：

case {'status': str(status), 'amount': float(amount)} if status == 'paid' and amount > 0:
    ...

守卫是模式匹配成功之后执行的布尔表达式，只有它为True时该分支才会被选中。这样就能在结构匹配的基础上叠加业务规则，而不会让模式本身变得冗长。

如果你的解析过程中需要复用一些复杂的匹配逻辑，还可以定义自定义匹配类。通过实现__match_args__和__eq__等方法，可以让自己的类支持模式匹配，但这超出本次案例范围，不再展开。

六、为什么match-case比字典映射更安全

有人可能会说，用一个大的字典把供应商特征映射到处理函数不是更灵活吗？类似于：

PARSERS = {
    'supplier_a': parse_a,
    'supplier_b': parse_b,
}
# 先通过某个特征找到键，再调用

这种做法的缺陷在于，你需要一种可靠的方式提取“特征键”，而这个提取过程本身就可能出错——比如不同供应商的type字段可能冲突，或者键的计算逻辑变得复杂。而模式匹配直接对整个数据结构进行形状识别，不需要显式的键分派，更能表达“这个数据长成什么样”的意图。此外，match-case在Python中是编译成比较高效的字节码的，对于频繁调用的解析函数来说，性能不会比手工if差。

七、一些实践中的小细节

使用match-case清洗JSON时，有几点经验值得分享：

• 总是放一个case _兜底。即便你觉得所有情况都覆盖了，上游的一个字段变动可能产生新的形状。兜底分支可以记录原始数据并抛出明确错误，防止静默丢弃。
• 注意字典键的匹配是按值还是按字面量。模式中{'status': str(status)}表示键名必须是字符串'status'，而{str(key): value}则表示匹配任意字符串键，并将键赋值给key。不要混淆。
• 在模式中使用类型提示不是强制的。写str(order_id)是模式语法，不是类型注解，它表示“匹配一个字符串值并将其绑定到变量order_id”。这和类型注解长得像但作用不同，不要误以为它会自动调用str()做类型转换。
• 尽量把复杂的模式写在前面。Python按从上到下的顺序尝试匹配case，所以把更具体、限制更多的模式放在前面，通用模式放在后面。