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
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]
RowGroup(b)
and add to a.nested_row_groups
:
b
has a parent
: a
RowGroup(a)
and add to b.parent
Thus RowGroup(a)
depends on RowGroup(a)
and you will recurse infinitely.
A few possible ways you can solve this:
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.
RowGroup
references: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