pycapnp

import capnp import os # Define a Cap'n Proto schema dynamically for demonstration # In a real application, this would be loaded from a .capnp file # e.g., addressbook = capnp.load('addressbook.capnp') SCHEMA_PATH = 'addressbook.capnp' addressbook_schema_content = ''' @0xd411d7353f406691; struct Person { id @0 :UInt32; name @1 :Text; email @2 :Text; phones @3 :List(PhoneNumber); struct PhoneNumber { number @0 :Text; type @1 :Type; enum Type { mobile @0; home @1; work @2; } } employment @4 :union { unemployed @5 :Void; employer @6 :Text; school @7 :Text; selfEmployed @8 :Void; } } struct AddressBook { people @0 :List(Person); } ''' # Write the schema to a temporary file with open(SCHEMA_PATH, 'w') as f: f.write(addressbook_schema_content) try: # Load the Cap'n Proto schema addressbook = capnp.load(SCHEMA_PATH) # 1. Build a message with capnp.alloc_builder() as msg_builder: addresses = msg_builder.init_root(addressbook.AddressBook) people = addresses.init('people', 2) alice = people[0] alice.id = 123 alice.name = 'Alice' alice.email = 'alice@example.com' alice_phones = alice.init('phones', 1) alice_phones[0].number = '555-1212' alice_phones[0].type = 'mobile' alice.employment.employer = 'Google' bob = people[1] bob.id = 456 bob.name = 'Bob' bob.email = 'bob@example.com' bob_phones = bob.init('phones', 2) bob_phones[0].number = '555-4567' bob_phones[0].type = 'home' bob_phones[1].number = '555-7654' bob_phones[1].type = 'work' bob.employment.unemployed = None # Serialize the message to bytes serialized_bytes = msg_builder.to_bytes_packed() print(f"Serialized message size: {len(serialized_bytes)} bytes") # 2. Read a message # Using capnp.alloc_reader() as a context manager is important for memory safety with capnp.alloc_reader(serialized_bytes) as msg_reader: read_addresses = msg_reader.get_root(addressbook.AddressBook) for person in read_addresses.people: print(f"\nPerson ID: {person.id}") print(f"Name: {person.name}") print(f"Email: {person.email}") print("Phones:") for phone in person.phones: print(f" - {phone.number} ({phone.type})") which_employment = person.employment.which() if which_employment == 'employer': print(f"Employment: Employer - {person.employment.employer}") elif which_employment == 'unemployed': print("Employment: Unemployed") else: print(f"Employment: {which_employment}") finally: # Clean up the temporary schema file if os.path.exists(SCHEMA_PATH): os.remove(SCHEMA_PATH)