GitLocker: The Coding Marketplace

Description:

prototopy 1.0.5

A Python package that

takes a str containing protobuf messages definitions
returns a types.ModuleType instance

It is useful for programs needing to en/decode protobuf messages for which the definition is provided as a string at runtime.
Installation
pip install proto-topy

Prerequisite: proto-topy needs protoc to be installed. On macOS, a simple brew install protobuf shall suffice.
single proto example: address book
Adaptation of the protocolbuffers example:
import requests
from pathlib import Path
from proto_topy import ProtoModule

# Retrieve protobuf messages definitions as a string
example_source = requests.get(
"https://raw.githubusercontent.com/protocolbuffers/protobuf/main/"
"examples/addressbook.proto").text

example_path = Path(
"protocolbuffers/protobuf/blob/main/examples/addressbook.proto")

# Compile and import
module = ProtoModule(file_path=example_path, source=example_source).compiled()

# Produce a serialized address book
address_book = module.py.AddressBook()
person = address_book.people.add()
person.id = 111
person.name = "A Name"
person.email = "a.name@mail.com"
phone_number = person.phones.add()
phone_number.number = "+1234567"
phone_number.type = module.py.Person.MOBILE
with open("address_book.data", "wb") as o:
o.write(address_book.SerializeToString())

# Use a serialized address book
address_book = module.py.AddressBook()
with open("address_book.data", "rb") as i:
address_book.ParseFromString(i.read())
for person in address_book.people:
print(person.id, person.name, person.email, phone_number.number)

multiple protos example
When several .proto need to be considered, use a ProtoCollection:
import sys
from pathlib import Path
from proto_topy import ProtoModule, ProtoCollection

module1 = ProtoModule(
file_path=Path("p1/p2/other2.proto"),
source="""
syntax = "proto3";
import "google/protobuf/timestamp.proto";
message OtherThing2 {
google.protobuf.Timestamp created = 1;
};"""
)

module2 = ProtoModule(
file_path=Path("p3/p4/test6.proto"),
source="""
syntax = "proto3";
import "p1/p2/other2.proto";
message Test6 {
OtherThing2 foo = 1;
};"""
)

collection = ProtoCollection(module1, module2).compiled()
sys.modules.update({proto.name: proto.py
for proto in collection.modules.values()})
print(sys.modules['test6'].Test6,
sys.modules['other2'].OtherThing2)

Stream of delimited messages
To decode a stream of contiguous protobuf messages of the same type, use DelimitedMessageFactory. Example:
from io import BytesIO
from pathlib import Path
from proto_topy import ProtoModule, DelimitedMessageFactory

# Generate Python module
module = ProtoModule(
file_path=Path("int32_streams.proto"),
source="""
syntax = "proto3";
message TestInt { int32 val = 1; };"""
).compiled()

# Feed a DelimitedMessageFactory with a sequence of TestInt instances for a range of 10 ints
integers = (module.py.TestInt(val=val) for val in range(10))
factory = DelimitedMessageFactory(BytesIO(), *integers)

# Rewind and read the stream of 10 protobuf messages
factory.rewind()
for offset_val in factory.message_read(module.py.TestInt):
print(f"TestInt message of val set to {offset_val[1]}")