python 验证“并行”JSON数组

lf3rwulv  于 2023-01-24  发布在  Python
关注(0)|答案(1)|浏览(142)

我尝试使用pydantic来验证以"并行"数组格式返回的JSON。也就是说，有一个定义列名/类型的数组，后面跟着一个"行"数组（这类似于panda处理df.to_json(orient='split')的方式，如here所示）

{
    "columns": [
        "sensor",
        "value",
        "range"
    ],
    "data": [
        [
            "a",
            1,
            {"low": 1, "high": 2}
        ],
        [
            "b",
            2,
            {"low": 0, "high": 2}
        ]
    ]
}

我知道我能做到：

class ValueRange(BaseModel):
    low: int
    high: int

class Response(BaseModel):
    columns: Tuple[Literal['sensor'], Literal['value'], Literal['range']]
    data: List[Tuple[str, int, ValueRange]]

但这也有一些缺点：

  • 解析之后，它不允许数据与列名的关联，因此，您必须通过索引来完成所有操作。理想情况下，我希望将响应解析为List[Row]，然后能够执行类似response.data[0].sensor的操作。
  • 它硬编码列顺序。
  • 它不允许响应中包含可变列。例如，同一端点还可以返回以下内容：
{
    "columns": ["sensor", "value"],
    "data": [
        ["a", 1],
        ["b", 2]
    ]
}

一开始我以为可以使用pydantic的区分联合，但我不知道如何跨数组执行此操作。
有人知道验证这种类型数据的最佳方法吗？（我目前使用pydantic，但如果有意义，我也可以使用其他库）。
谢谢!

python

1条答案

按热度按时间
vd8tlhqk

vd8tlhqk1#

TL; DR

custom validator的一个非常有趣的用例。

from collections.abc import Sequence
from typing import Literal, Optional

from pydantic import BaseModel, validator
from pydantic.fields import Field

Column = Literal["sensor", "value", "range"]

class ValueRange(BaseModel):
    low: int
    high: int

class DataPoint(BaseModel):
    sensor: str
    value: int
    range: Optional[ValueRange]

class Response(BaseModel):
    columns: Optional[tuple[Column, ...]] = Field(exclude=True, repr=False)
    data: list[DataPoint]

    @validator("columns", pre=True)
    def ensure_distinct(cls, v: object) -> object:
        if isinstance(v, Sequence) and len(v) != len(set(v)):
            raise ValueError("`columns` must all be distinct")
        return v

    @validator("data", pre=True, each_item=True)
    def parse_nested_sequence(
        cls,
        v: object,
        values: dict[str, object],
    ) -> object:
        if not isinstance(v, Sequence):
            return v
        columns = values.get("columns")
        if not isinstance(columns, Sequence):
            raise TypeError(
                "If `data` items are provided as a sequences, "
                "the `columns` must be present as a sequence."
            )
        if len(columns) != len(v):
            raise ValueError(
                "`data` item must be the same length as `columns`"
            )
        return dict(zip(columns, v))

解释

架构

首先，我们需要设置模型，以反映我们希望在解析和验证完成后拥有的模式。
由于您提到您希望响应模型中的data字段是对应于某个模式的模型示例列表，因此我们需要定义该模式。从您的示例来看，似乎至少需要sensor字段。value和range。range字段应该再次成为它自己的模型。您还提到range应该是可选的，所以我们也将对其进行编码。
实际的顶层Response模型仍然会有columns字段，因为我们需要它来进行验证，但是我们可以像dict和json一样，在字符串表示和exporter methods中隐藏该字段，为了在columns元组中传递可变数量的元素，我们将稍微修改注解。
以下是我建议的模式：

from collections.abc import Sequence
from typing import Literal, Optional

from pydantic import BaseModel, validator
from pydantic.fields import Field

Column = Literal["sensor", "value", "range"]

class ValueRange(BaseModel):
    low: int
    high: int

class DataPoint(BaseModel):
    sensor: str
    value: int
    range: Optional[ValueRange]

class Response(BaseModel):
    columns: Optional[tuple[Column, ...]] = Field(exclude=True, repr=False)
    data: list[DataPoint]

    ...  # more code

字段类型和参数

要隐藏columns，我们可以使用Field构造函数的适当参数，将其设置为Optional（默认值为None）意味着我们仍然可以在没有它的情况下初始化Response的示例，但是我们当然会被迫以正确的格式提供data列表（作为字典或DataPoint示例）。
因为我们使用tuple[Column, ...]作为columns的注解，所以元素可以是任何顺序，这正是您想要的，但理论上它也可以是任意长度的，并且包含一堆重复项，Python类型系统没有提供任何优雅的工具来定义类型，以指示元组的所有元素必须是不同的。当然，我们可以构造一个包含所有应该有效的字面值排列的大型类型联合，但这几乎不切实际。

验证器

相反，我建议使用一个非常简单的验证器来执行此检查：

...

class Response(BaseModel):
    columns: Optional[tuple[Column, ...]] = Field(exclude=True, repr=False)
    data: list[DataPoint]

    @validator("columns", pre=True)
    def ensure_distinct(cls, v: object) -> object:
        if isinstance(v, Sequence) and len(v) != len(set(v)):
            raise ValueError("`columns` must all be distinct")
        return v

    ...  # more code

pre=True在这里实际上很重要，但只是与第二个验证器一起使用，第二个验证器更有趣，因为它应该将这些列与数据序列放在一起，下面是我的建议：

...

class Response(BaseModel):
    columns: Optional[tuple[Column, ...]] = Field(exclude=True, repr=False)
    data: list[DataPoint]

    @validator("columns", pre=True)
    def ensure_distinct(cls, v: object) -> object:
        if isinstance(v, Sequence) and len(v) != len(set(v)):
            raise ValueError("`columns` must all be distinct")
        return v

    @validator("data", pre=True, each_item=True)
    def parse_nested_sequence(
        cls,
        v: object,
        values: dict[str, object],
    ) -> object:
        if not isinstance(v, Sequence):
            return v
        columns = values.get("columns")
        if not isinstance(columns, Sequence):
            raise TypeError(
                "If `data` items are provided as a sequences, "
                "the `columns` must be present as a sequence."
            )
        if len(columns) != len(v):
            raise ValueError(
                "`data` item must be the same length as `columns`"
            )
        return dict(zip(columns, v))

两个验证器上的pre=True确保它们在这些字段类型的默认验证器之前运行（否则我们将立即从示例data中得到验证错误）。字段验证器总是按照字段定义的顺序调用**，因此我们确保在调用自定义data验证器之前调用自定义columns验证器。
这个顺序还允许我们在data验证器的values字典中访问columns验证器的输出。
each_item=True标志改变了验证器的行为，使其应用于data列表的每个元素，而不是整个列表。这意味着对于我们的示例数据，v参数将始终是一个"子列表"（例如["a", 1]）。
如果要验证的值不是sequence类型，我们就不去管它了，它将由默认的字段验证器进行适当的处理。如果它 * 是 * 一个sequence，我们需要确保columns存在，也是一个sequence，并且它们具有相同的长度。如果这些检查通过，我们就可以压缩它们。将zip压缩到字典中，然后轻松地将其发送到默认字段验证器。
就这样。

演示

下面是一个小的演示脚本：

def main() -> None:
    from pydantic import ValidationError
    print(Response.parse_raw(TEST_JSON_VALID_1).json(indent=2), "\n")
    print(Response.parse_raw(TEST_JSON_VALID_2).json(indent=2), "\n")
    try:
        Response.parse_raw(TEST_JSON_INVALID_1)
    except ValidationError as exc:
        print(exc.json(indent=2), "\n")
    try:
        Response.parse_raw(TEST_JSON_INVALID_2)
    except ValidationError as exc:
        print(exc.json(indent=2), "\n")
    try:
        Response.parse_raw(TEST_JSON_INVALID_3)
    except ValidationError as exc:
        print(exc.json(indent=2))

if __name__ == "__main__":
    main()

下面是测试数据及其相应输出：

TEST_JSON_VALID_1
{
    "columns": [
        "sensor",
        "value",
        "range"
    ],
    "data": [
        [
            "a",
            1,
            {"low": 1, "high": 2}
        ],
        [
            "b",
            2,
            {"low": 0, "high": 2}
        ]
    ]
}
{
  "data": [
    {
      "sensor": "a",
      "value": 1,
      "range": {
        "low": 1,
        "high": 2
      }
    },
    {
      "sensor": "b",
      "value": 2,
      "range": {
        "low": 0,
        "high": 2
      }
    }
  ]
}
TEST_JSON_VALID_2（订单不同且无range）
{
    "columns": ["value", "sensor"],
    "data": [
        [1, "a"],
        [2, "b"]
    ]
}
{
  "data": [
    {
      "sensor": "a",
      "value": 1,
      "range": null
    },
    {
      "sensor": "b",
      "value": 2,
      "range": null
    }
  ]
}
TEST_JSON_INVALID_1
{
    "columns": ["foo", "value"],
    "data": []
}
[
  {
    "loc": [
      "columns",
      0
    ],
    "msg": "unexpected value; permitted: 'sensor', 'value', 'range'",
    "type": "value_error.const",
    "ctx": {
      "given": "foo",
      "permitted": [
        "sensor",
        "value",
        "range"
      ]
    }
  }
]
TEST_JSON_INVALID_2

一个一个一个一个

TEST_JSON_INVALID_3

一个一个三个一个一个

赞(0）分享  回复(0）举报  2023-01-24
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com