如何实现Pydantic中MCC与type字段的关联验证逻辑？

问题需求 #

需要实现两个字段的关联验证逻辑：

• 当MCC字段不为空时，type字段必须传入OUTSIDE
• 当MCC字段为空时，type字段必须传入INSIDE

原代码错误原因 #

原代码使用@validator('type', 'MCC')是错误用法：普通validator每次仅处理单个字段，无法同时获取两个字段的值做关联校验。多字段关联验证必须使用root_validator，它能获取所有已完成基础验证的字段值字典。

正确实现代码 #

import json

from pydantic.dataclasses import dataclass, ValidationError
from pydantic import root_validator
from typing import Union, List, Literal


@dataclass
class DataInclude:
    type: Literal['INSIDE', 'OUTSIDE']
    accountID: Union[None, int]
    date: int
    tranDate: int
    operationType: Literal['CREDIT', 'DEBIT', 'OPEN', 'DV']
    paymentDetailType: Literal[
        'BETWEEN_THEIR', 'INSIDE_BANK', 'EXTERNAL_INDIVIDUAL', 'EXTERNAL_ENTITY', 'OTHER_BANK',
        'HOUSING_AND_COMMUNAL_SERVICE', 'MOBILE', 'INTERNET', 'TRANSPORT', 'TAX_AND_STATE_SERVICE',
        'NOT_FINANCE', 'CONTACT_ADDRESSLESS', 'DIRECT', 'SFP', 'OUTSIDE_CASH', 'INSIDE_OTHER',
        'OUTSIDE_OTHER', 'C2B_PAYMENT', 'INSIDE_DEPOSIT']
    amount: Union[int, float, None]
    documentAmount: Union[int, float, None]
    comment: str
    documentID: int | None
    accountNumber: str
    currencyCodeNumeric: int
    merchantName: str | None
    merchantNameRus: str | None
    groupName: str
    md5hash: str
    svgImage: str | None
    fastPayment: str | None
    terminalCode: str | None
    deviceCode: str | None
    country: str | None
    city: str | None
    operationId: str | None
    isCancellation: bool | None  # BOOL!
    cardTranNumber: str | None
    opCode: int | None
    MCC: int | None
    
    @root_validator(skip_on_failure=True)
    def check_type_and_mcc(cls, values):
        type_op = values.get('type')
        mcc = values.get('MCC')
        
        # 验证MCC非空时type必须为OUTSIDE
        if mcc is not None and type_op != 'OUTSIDE':
            raise ValueError('当MCC不为空时，type必须为OUTSIDE')
        # 验证MCC为空时type必须为INSIDE
        if mcc is None and type_op != 'INSIDE':
            raise ValueError('当MCC为空时，type必须为INSIDE')
        
        return values


@dataclass
class MessageResponse:
    statusCode: int
    errorMessage: Union[None, str]
    data: List[DataInclude]

    @staticmethod
    def validation_body(data):
        try:
            data_new = json.loads(data)
            MessageResponse(**data_new)
            return True
        except ValidationError as e:
            raise e

关键修改说明 #

1. 替换验证器类型：使用@root_validator(skip_on_failure=True)替代原@validator，skip_on_failure=True表示当其他字段验证失败时，跳过当前关联校验，避免抛出无关错误。
2. 双向逻辑校验：同时覆盖两种场景的验证规则，确保type和MCC的关联关系完全符合需求。
3. 安全取值：用values.get()获取字段值，避免因字段未通过前置验证导致的KeyError。

内容的提问来源于stack exchange，提问作者Alex