def get_proto_data()

in components/doc-registry/src/document_registry_service.py [0:0]

• 15 lines of code
• 4 McCabe index (conditional complexity)

def get_proto_data(obj: Sequence[proto.Message], with_schema: bool = True):
    """Convert a sequence of messages into proto data"""

    proto_data = types.AppendRowsRequest.ProtoData()

    # Bring in the schema if requested (required first time)
    if with_schema:
        proto_schema = types.ProtoSchema()
        proto_descriptor = descriptor_pb2.DescriptorProto()  # pylint: disable=no-member
        type(obj[0]).pb().DESCRIPTOR.CopyToProto(proto_descriptor)
        proto_schema.proto_descriptor = proto_descriptor
        proto_data.writer_schema = proto_schema

    # Serialize the rows
    proto_rows = types.ProtoRows()
    for o in obj:
        proto_rows.serialized_rows.append(
            type(o).serialize(o)
        )  # pylint: disable=no-member

    proto_data.rows = proto_rows

    return proto_data