Recursive Pydantic model to gRPC protobuf

huangapple go评论62阅读模式
英文:

Recursive Pydantic model to gRPC protobuf

问题

可以将递归的 pydantic 模型转换为 protobuf 并通过 gRPC 发送吗?

示例:

from __future__ import annotations

from typing import Optional

from pydantic import BaseModel

class RowGroup(BaseModel):
    report_code: Optional[str]
    tab: Optional[str]
    parent: Optional[RowGroup]
    nested_row_groups: Optional[list[RowGroup]]

Proto 消息:

message RowGroup {
    string report_code = 1;
    string tab = 2;
    RowGroup parent = 3;
    repeated RowGroup nested_row_groups = 4;
}

填充数据:

def get_recursion_scenario() -> RowGroup:
    a = RowGroup(report_code='a', tab='a tab', nested_row_groups=None, parent=None)
    b = RowGroup(report_code='b', tab='b tab', parent=a, nested_row_groups=None)
    c = RowGroup(report_code='c', tab='c tab', parent=a, nested_row_groups=None)
    a.nested_row_groups = [b, c]
    return a

如果我尝试这样转换,我的程序会因递归深度限制而失败:

def to_proto(pydantic_a):
    if pydantic_a is None:
        return None
    grpc_model = fill_row_groups_pb2.RowGroup(
        tab=pydantic_a.tab,
        report_code=pydantic_a.report_code,
        parent=to_proto(pydantic_a.parent),
        nested_row_groups=list(to_proto(i) for i in pydantic_a.nested_row_groups)
    )
    return grpc_model
英文:

Is it possible to convert recursive pydantic model to protobuf and send it through gRPC?

Example:

from __future__ import annotations

from typing import Optional

from pydantic import BaseModel

class RowGroup(BaseModel):
    report_code: Optional[str]
    tab: Optional[str]
    parent: Optional[RowGroup]
    nested_row_groups: Optional[list[RowGroup]]

Proto message:

message RowGroup {
    string report_code = 1;
    string tab = 2;
    RowGroup parent = 3;
    repeated RowGroup nested_row_groups = 4;
}

Filling data:

def get_recursion_scenario() -> RowGroup:
    a = RowGroup(report_code='a', tab='a tab', nested_row_groups=None, parent=None)
    b = RowGroup(report_code='b', tab='b tab', parent=a, nested_row_groups=None)
    c = RowGroup(report_code='c', tab='c tab', parent=a, nested_row_groups=None)
    a.nested_row_groups = [b, c]
    return a

If I try convert like this, my program fails from recursion depth limit:

def to_proto(pydantic_a):
    if pydantic_a is None:
        return None
    grpc_model = fill_row_groups_pb2.RowGroup(
        tab=pydantic_a.tab,
        report_code=pydantic_a.report_code,
        parent=to_proto(pydantic_a.parent),
        nested_row_groups=list(to_proto(i) for i in pydantic_a.nested_row_groups)
    )
    return grpc_model

答案1

得分: 1

你的问题在于子节点上的 parent 属性的解析。

模型 a 有两个子节点(例如 nested_row_groups),bc

  a
 / \
b   c

观察到以下无限循环:

在实例化对象 RowGroup(a) 时:

  • a 有 2 个子节点 nested_row_groups[b, c]
  • 实例化 RowGroup(b) 并添加到 a.nested_row_groups
    • b 有一个 parenta
      • 实例化 RowGroup(a) 并添加到 b.parent

因此 RowGroup(a) 依赖于 RowGroup(a),并将导致无限递归。

可能的解决方案:

有几种可能的解决方法:

1. 将 parent 存储为 id 而不是对象:

如果 parent 属性是一个标识符(id),而不是一个完整的对象,就不会发生无限递归。这将需要对您的协议缩写定义以及 pydantic 模型进行更改。

2. 取消链接 RowGroup 引用:

创建一个没有引用的 RowItem 对象,然后创建一个链接这两个对象的 RowGroup 对象:(我建议采用这种方法...)

Proto:

message RowItem {
    string report_code = 1;
    string tab = 2;
}

message RowGroup {
    RowItem item = 1;
    RowItem parent = 3;
    repeated RowGroup nested_row_groups = 4;
}

您的 pydantic 模型将如下所示:

class RowItem(BaseModel):
    report_code: Optional[str]
    tab: Optional[str]

class RowGroup(BaseModel):
    item: RowItem
    parent: Optional[RowItem]
    nested_row_groups: Optional[list[RowGroup]]

功能代码:

def get_recursion_scenario() -> RowGroup:
    a = RowGroup(
            item=RowItem(report_code='a', tab='a tab')
        )
    b = RowGroup(
            item=RowItem(report_code='b', tab='b tab'),
            parent=a
        )
    c = RowGroup(
            item=RowItem(report_code='c', tab='c tab'),
            parent=a
        )
    a.nested_row_groups = [b, c]
    return a
英文:

Your issue here is the resolution of the parent attribute on the child nodes.

Model a has 2 children (e.g. nested_row_groups), b and c:

  a
 / \
b   c

The following infinite cycle is observed:

When instantiating your object RowGroup(a):

  • a has 2 children nested_row_groups: [b, c]
  • Instantiate RowGroup(b) and add to a.nested_row_groups:
    • b has a parent: a
      • Instantiate RowGroup(a) and add to b.parent

Thus RowGroup(a) depends on RowGroup(a) and you will recurse infinitely.

Possible solutions:

A few possible ways you can solve this:

1. Store parent as an id instead of an object:

If the parent attribute was an id, instead of an entire object, you wouldn't have infinite recursion. This would require changes to your protobuf definition as well as pydantic models.

Create a RowItem object without references, then create a RowGroup object linking the two: (I recommend doing this...)

Proto:

message RowItem {
    string report_code = 1;
    string tab = 2;
}

message RowGroup {
    RowItem item = 1;
    RowItem parent = 3;
    repeated RowGroup nested_row_groups = 4;
}

Your pydantic models would look like:

class RowItem(BaseModel):
    report_code: Optional[str]
    tab: Optional[str]

class RowGroup(BaseModel):
    item: RowItem
    parent: Optional[RowItem]
    nested_row_groups: Optional[list[RowGroup]]

Functional code:

def get_recursion_scenario() -> RowGroup:
    a = RowGroup(
            item=RowItem(report_code='a', tab='a tab')
        )
    b = RowGroup(
            item=RowItem(report_code='b', tab='b tab'),
            parent=a
        )
    c = RowGroup(
            item=RowItem(report_code='c', tab='c tab'),
            parent=a
        )
    a.nested_row_groups = [b, c]
    return a

huangapple
  • 本文由 发表于 2023年3月31日 22:11:07
  • 转载请务必保留本文链接:https://go.coder-hub.com/75899510.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定