pycapnp_schema_definition_and_message_read_write.py

1import capnp
2import os
3
4# First, we need to create a schema file for this example
5schema_content = """
6@0xdbb9ad1f1444f837;  # unique file ID
7
8struct Person {
9  id @0 :UInt32;
10  name @1 :Text;
11  email @2 :Text;
12  phones @3 :List(PhoneNumber);
13
14  struct PhoneNumber {
15    number @0 :Text;
16    type @1 :Type;
17
18    enum Type {
19      mobile @0;
20      home @1;
21      work @2;
22    }
23  }
24
25  employment :union {
26    unemployed @4 :Void;
27    employer @5 :Text;
28    school @6 :Text;
29    selfEmployed @7 :Void;
30  }
31}
32
33struct AddressBook {
34  people @0 :List(Person);
35}
36"""
37
38with open('addressbook.capnp', 'w') as f:
39    f.write(schema_content)
40
41# Load the schema
42addressbook_capnp = capnp.load('addressbook.capnp')
43
44def write_address_book(file):
45    address_book = addressbook_capnp.AddressBook.new_message()
46    people = address_book.init('people', 2)
47
48    alice = people[0]
49    alice.id = 123
50    alice.name = 'Alice'
51    alice.email = 'alice@example.com'
52    alice_phones = alice.init('phones', 1)
53    alice_phones[0].number = '555-1212'
54    alice_phones[0].type = 'mobile'
55    alice.employment.school = 'MIT'
56
57    bob = people[1]
58    bob.id = 456
59    bob.name = 'Bob'
60    bob.email = 'bob@example.com'
61    bob_phones = bob.init('phones', 2)
62    bob_phones[0].number = '555-4567'
63    bob_phones[0].type = 'home'
64    bob_phones[1].number = '555-7654'
65    bob_phones[1].type = 'work'
66    bob.employment.unemployed = None
67
68    address_book.write(file)
69
70def print_address_book(file):
71    address_book = addressbook_capnp.AddressBook.read(file)
72
73    for person in address_book.people:
74        print(f'{person.name}: {person.email}')
75        for phone in person.phones:
76            print(f'  {phone.type} phone: {phone.number}')
77
78        employment = person.employment
79        if employment.which() == 'unemployed':
80            print('  unemployed')
81        elif employment.which() == 'employer':
82            print(f'  employer: {employment.employer}')
83        elif employment.which() == 'school':
84            print(f'  student at: {employment.school}')
85        elif employment.which() == 'selfEmployed':
86            print('  self-employed')
87
88if __name__ == '__main__':
89    with open('addressbook.bin', 'w+b') as f:
90        write_address_book(f)
91        f.seek(0)
92        print_address_book(f)
93    
94    # Clean up
95    os.remove('addressbook.capnp')
96    os.remove('addressbook.bin')