Search code examples
pythonprotocol-buffersgrpcpydantic

Recursive Pydantic model to gRPC protobuf


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

Solution

  • 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.

    2. Unlink 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