Pydantic 对我的类型及其联合感到困惑

Question

我有以下 Pydantic 模型类型方案规范：

class RequestPayloadPositionsParams(BaseModel):
    """
    Request payload positions parameters
    """

    account: str = Field(default="COMBINED ACCOUNT")
    fields: List[str] = Field(default=["QUANTITY", "OPEN_PRICE", "OPEN_COST"])


class RequestPayloadPositions(BaseModel):
    """
    Request payload positions service
    """

    header: RequestPayloadHeader = Field(
        default=RequestPayloadHeader(service="positions", id="positions", ver=0)
    )
    params: RequestPayloadPositionsParams = Field(
        default=RequestPayloadPositionsParams()
    )


class RequestPayloadOrdersParams(BaseModel):
    """
    Request payload orders parameters
    """

    account: str = Field(default="COMBINED ACCOUNT")
    types: List[str] = Field(default=["WORKING", "FILLED", "CANCELED"])


class RequestPayloadOrders(BaseModel):
    """
    Request payload orders service
    """

    header: RequestPayloadHeader = Field(
        default=RequestPayloadHeader(service="order_events", id="order_events", ver=0)
    )
    params: RequestPayloadOrdersParams = Field(default=RequestPayloadOrdersParams())


class RequestPayload(BaseModel):
    """
    Request payload data
    """

    payload: List[Union[RequestPayloadPositions, RequestPayloadOrders]] = Field(...)

现在，我想为订单和仓位服务创建一个有效负载对象：

positions = requests.RequestPayload(payload=[requests.RequestPayloadPositions()])
orders = requests.RequestPayload(payload=[requests.RequestPayloadOrders()])

现在，positions有类型requests.RequestPayload[payload= requests.RequestPayloadPositions...但orders没有requests.RequestPayload[payload= requests.RequestPayloadOrders但相同positions。这是错误的。

我可以通过将模型规范从 更改payload: List[Union[RequestPayloadPositions, RequestPayloadOrders]] = Field(...)为payload: List[Any] = Field(...)... 来解决此问题，但我想明确定义允许的类型。

知道如何解决这个问题，还是我应该解释得更详细？你明白我的问题吗？

编辑 工作代码示例，显示最后一行中的第二个断言失败，但不应失败......

from typing import List, Union
from pydantic import BaseModel, Field


class RequestPayloadHeader(BaseModel):
    """
    Request payload header
    """

    service: str = Field(...)
    id: str = Field(...)
    ver: int = Field(...)


class RequestPayloadLoginParams(BaseModel):
    """
    Request payload login parameters
    """

    domain: str = Field(default="TOS")
    platform: str = Field(default="PROD")
    token: str = Field(...)
    accessToken: str = Field(default="")
    tag: str = Field(default="TOSWeb")


class RequestPayloadLogin(BaseModel):
    """
    Request payload login service
    """

    header: RequestPayloadHeader = Field(
        default=RequestPayloadHeader(service="login", id="login", ver=0)
    )
    params: RequestPayloadLoginParams = Field(...)


class RequestPayloadPositionsParams(BaseModel):
    """
    Request payload positions parameters
    """

    account: str = Field(default="COMBINED ACCOUNT")
    fields: List[str] = Field(default=["QUANTITY", "OPEN_PRICE", "OPEN_COST"])


class RequestPayloadOrdersParams(BaseModel):
    """
    Request payload orders parameters
    """

    account: str = Field(default="COMBINED ACCOUNT")
    types: List[str] = Field(default=["WORKING", "FILLED", "CANCELED"])


class RequestPayloadService(BaseModel):
    """
    Request payload service
    """

    header: RequestPayloadHeader = Field(...)
    params: Union[RequestPayloadPositionsParams, RequestPayloadOrdersParams] = Field(
        ...
    )


class RequestPayload(BaseModel):
    """
    Request payload data
    """

    payload: List[Union[RequestPayloadLogin, RequestPayloadService]] = Field(...)


if __name__ == "__main__":
    positions = RequestPayload(
        payload=[
            RequestPayloadService(
                header=RequestPayloadHeader(service="positions", id="positions", ver=0),
                params=RequestPayloadPositionsParams(),
            )
        ]
    )
    assert isinstance(positions.payload[0].params, RequestPayloadPositionsParams)
    orders = RequestPayload(
        payload=[
            RequestPayloadService(
                header=RequestPayloadHeader(
                    service="order_events", id="order_events", ver=0
                ),
                params=RequestPayloadOrdersParams(),
            )
        ]
    )
    assert isinstance(orders.payload[0].params, RequestPayloadOrdersParams)

EDIT2 Alex 的解决方案不涵盖当我有两个具有相同字段名称但类型不同的模型时的场景，如下所示：

class ResponseReplacePatchStr(BaseModel):
    op: str = Field(default="replace")
    path: str = Field(...)
    value: str = Field(...)

    class Config:
        extra = "forbid"

class ResponseReplacePatchFloat(BaseModel):
    op: str = Field(default="replace")
    path: str = Field(...)
    value: float = Field(...)

    class Config:
        extra = "forbid"

如果是第一个类型，则它始终会转换为字段str的类型valueResponseReplacePatchStr

Union[
            ResponseReplacePatchStr, ResponseReplacePatchFloat
        ]

我怎样才能解决这个问题，以便 Pydantic 照顾我的类型？

Answer 1

这是 Pydantic 匹配的特点之一Union，其描述为：

然而，如上所示，pydantic 将尝试“匹配”Union 下定义的任何类型，并将使用第一个匹配的类型。[...]

因此，建议在定义联合注释时，首先包含最具体的类型，然后包含不太具体的类型。

同时，默认情况下会忽略额外的字段，并在您的情况下使用声明字段的默认值。

因此，解决方案可能是添加extra = 'forbid'模型配置选项：

class RequestPayloadPositionsParams(BaseModel):
    """
    Request payload positions parameters
    """
    account: str = Field(default="COMBINED ACCOUNT")
    fields: List[str] = Field(default=["QUANTITY", "OPEN_PRICE", "OPEN_COST"])

    class Config:
        extra = 'forbid'



class RequestPayloadOrdersParams(BaseModel):
    """
    Request payload orders parameters
    """
    account: str = Field(default="COMBINED ACCOUNT")
    types: List[str] = Field(default=["WORKING", "FILLED", "CANCELED"])

    class Config:
        extra = 'forbid'