Pydantic 对我的类型和他们的联合感到困惑

Pydantic is confused with my types and their union

我有以下 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"

如果 ResponseReplacePatchStr

中提到的第一个类型,则 value 字段总是转换为类型 str
Union[
            ResponseReplacePatchStr, ResponseReplacePatchFloat
        ]

我怎样才能解决这个问题,让 Pydantic 照顾我的类型?

这是 Pydantic 匹配的功能之一 Union,即 described 如:

However, as can be seen above, pydantic will attempt to 'match' any of the types defined under Union and will use the first one that matches.[...]

As such, it is recommended that, when defining Union annotations, the most specific type is included first and followed by less specific types.

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

因此,解决方案可能是添加 extra = 'forbid' 模型 config 选项:

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'