Recursive Pydantic model to gRPC protobuf

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

Recursive Pydantic model to gRPC protobuf

问题

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

示例：

  1. from __future__ import annotations
  3. from typing import Optional
  5. from pydantic import BaseModel
  7. class RowGroup(BaseModel):
  8.     report_code: Optional[str]
  9.     tab: Optional[str]
  10.     parent: Optional[RowGroup]
  11.     nested_row_groups: Optional[list[RowGroup]]

Proto 消息：

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

填充数据：

  1. def get_recursion_scenario() -> RowGroup:
  2.     a = RowGroup(report_code='a', tab='a tab', nested_row_groups=None, parent=None)
  3.     b = RowGroup(report_code='b', tab='b tab', parent=a, nested_row_groups=None)
  4.     c = RowGroup(report_code='c', tab='c tab', parent=a, nested_row_groups=None)
  5.     a.nested_row_groups = [b, c]
  6.     return a

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

  1. def to_proto(pydantic_a):
  2.     if pydantic_a is None:
  3.         return None
  4.     grpc_model = fill_row_groups_pb2.RowGroup(
  5.         tab=pydantic_a.tab,
  6.         report_code=pydantic_a.report_code,
  7.         parent=to_proto(pydantic_a.parent),
  8.         nested_row_groups=list(to_proto(i) for i in pydantic_a.nested_row_groups)
  9.     )
  10.     return grpc_model
英文:

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

Example:

  1. from __future__ import annotations
  3. from typing import Optional
  5. from pydantic import BaseModel
  7. class RowGroup(BaseModel):
  8.     report_code: Optional[str]
  9.     tab: Optional[str]
  10.     parent: Optional[RowGroup]
  11.     nested_row_groups: Optional[list[RowGroup]]

Proto message:

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

Filling data:

  1. def get_recursion_scenario() -> RowGroup:
  2.     a = RowGroup(report_code='a', tab='a tab', nested_row_groups=None, parent=None)
  3.     b = RowGroup(report_code='b', tab='b tab', parent=a, nested_row_groups=None)
  4.     c = RowGroup(report_code='c', tab='c tab', parent=a, nested_row_groups=None)
  5.     a.nested_row_groups = [b, c]
  6.     return a

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

  1. def to_proto(pydantic_a):
  2.     if pydantic_a is None:
  3.         return None
  4.     grpc_model = fill_row_groups_pb2.RowGroup(
  5.         tab=pydantic_a.tab,
  6.         report_code=pydantic_a.report_code,
  7.         parent=to_proto(pydantic_a.parent),
  8.         nested_row_groups=list(to_proto(i) for i in pydantic_a.nested_row_groups)
  9.     )
  10.     return grpc_model

答案1

得分: 1

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

模型 a 有两个子节点（例如 nested_row_groups），b 和 c：

  1.   a
  2.  / \
  3. b   c

观察到以下无限循环：

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

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

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

可能的解决方案：

有几种可能的解决方法：

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

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

2. 取消链接 RowGroup 引用：

创建一个没有引用的 RowItem 对象，然后创建一个链接这两个对象的 RowGroup 对象：（我建议采用这种方法...）

Proto:

  1. message RowItem {
  2.     string report_code = 1;
  3.     string tab = 2;
  4. }
  6. message RowGroup {
  7.     RowItem item = 1;
  8.     RowItem parent = 3;
  9.     repeated RowGroup nested_row_groups = 4;
  10. }

您的 pydantic 模型将如下所示：

  1. class RowItem(BaseModel):
  2.     report_code: Optional[str]
  3.     tab: Optional[str]
  5. class RowGroup(BaseModel):
  6.     item: RowItem
  7.     parent: Optional[RowItem]
  8.     nested_row_groups: Optional[list[RowGroup]]

功能代码：

  1. def get_recursion_scenario() -> RowGroup:
  2.     a = RowGroup(
  3.             item=RowItem(report_code='a', tab='a tab')
  4.         )
  5.     b = RowGroup(
  6.             item=RowItem(report_code='b', tab='b tab'),
  7.             parent=a
  8.         )
  9.     c = RowGroup(
  10.             item=RowItem(report_code='c', tab='c tab'),
  11.             parent=a
  12.         )
  13.     a.nested_row_groups = [b, c]
  14.     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:

  1.   a
  2.  / \
  3. 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:

  1. message RowItem {
  2.     string report_code = 1;
  3.     string tab = 2;
  4. }
  6. message RowGroup {
  7.     RowItem item = 1;
  8.     RowItem parent = 3;
  9.     repeated RowGroup nested_row_groups = 4;
  10. }

Your pydantic models would look like:

  1. class RowItem(BaseModel):
  2.     report_code: Optional[str]
  3.     tab: Optional[str]
  5. class RowGroup(BaseModel):
  6.     item: RowItem
  7.     parent: Optional[RowItem]
  8.     nested_row_groups: Optional[list[RowGroup]]

Functional code:

  1. def get_recursion_scenario() -> RowGroup:
  2.     a = RowGroup(
  3.             item=RowItem(report_code='a', tab='a tab')
  4.         )
  5.     b = RowGroup(
  6.             item=RowItem(report_code='b', tab='b tab'),
  7.             parent=a
  8.         )
  9.     c = RowGroup(
  10.             item=RowItem(report_code='c', tab='c tab'),
  11.             parent=a
  12.         )
  13.     a.nested_row_groups = [b, c]
  14.     return a

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

发表评论

匿名网友

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

确定