0
# Protocol Buffer Helpers
1
2
Helper functions for working with Google's protocol buffer messages, including type conversion, field manipulation, and message introspection. These utilities provide consistent interfaces for handling protobuf messages across different Google APIs.
3
4
## Capabilities
5
6
### Message Type Utilities
7
8
Functions for working with protobuf message types and Any messages.
9
10
```python { .api }
11
def from_any_pb(pb_type, any_pb):
12
"""
13
Converts an Any protobuf message to the specified message type.
14
15
Args:
16
pb_type (type): Target protobuf message class
17
any_pb: google.protobuf.any_pb2.Any message
18
19
Returns:
20
Message: Deserialized message of specified type
21
22
Raises:
23
TypeError: If any_pb is not compatible with pb_type
24
"""
25
26
def get_messages(module):
27
"""
28
Discover all protobuf Message classes in a module.
29
30
Args:
31
module: Python module containing protobuf classes
32
33
Returns:
34
dict: Mapping of message names to message classes
35
"""
36
37
def check_oneof(**kwargs):
38
"""
39
Validate that exactly one of the keyword arguments is set.
40
41
Args:
42
**kwargs: Keyword arguments to validate
43
44
Raises:
45
TypeError: If zero or more than one argument is provided
46
47
Returns:
48
tuple: (key, value) of the single provided argument
49
"""
50
51
def field_mask(original, modified):
52
"""
53
Create a field mask by comparing two protobuf messages.
54
55
Args:
56
original: Original protobuf message
57
modified: Modified protobuf message
58
59
Returns:
60
google.protobuf.field_mask_pb2.FieldMask: Field mask representing changes
61
"""
62
```
63
64
### Message Field Manipulation
65
66
Functions for getting, setting, and manipulating fields in protobuf messages and dictionaries.
67
68
```python { .api }
69
def get(msg_or_dict, key, default=None):
70
"""
71
Get a value from a protobuf Message or dictionary.
72
73
Args:
74
msg_or_dict: Protobuf message or dictionary
75
key (str): Field name or dictionary key
76
default: Default value if key not found
77
78
Returns:
79
Any: Field value or default
80
"""
81
82
def set(msg_or_dict, key, value):
83
"""
84
Set a value on a protobuf Message or dictionary.
85
86
Args:
87
msg_or_dict: Protobuf message or dictionary
88
key (str): Field name or dictionary key
89
value: Value to set
90
"""
91
92
def setdefault(msg_or_dict, key, value):
93
"""
94
Set a value only if the current value is falsy.
95
96
Args:
97
msg_or_dict: Protobuf message or dictionary
98
key (str): Field name or dictionary key
99
value: Value to set if current value is falsy
100
101
Returns:
102
Any: Current value (existing or newly set)
103
"""
104
```
105
106
### Type Constants
107
108
Constants representing common protobuf wrapper types.
109
110
```python { .api }
111
# Tuple of protobuf wrapper types
112
_WRAPPER_TYPES = (
113
"google.protobuf.wrappers_pb2.BoolValue",
114
"google.protobuf.wrappers_pb2.BytesValue",
115
"google.protobuf.wrappers_pb2.DoubleValue",
116
"google.protobuf.wrappers_pb2.FloatValue",
117
"google.protobuf.wrappers_pb2.Int32Value",
118
"google.protobuf.wrappers_pb2.Int64Value",
119
"google.protobuf.wrappers_pb2.StringValue",
120
"google.protobuf.wrappers_pb2.UInt32Value",
121
"google.protobuf.wrappers_pb2.UInt64Value"
122
)
123
124
# Sentinel object for default values
125
_SENTINEL = object()
126
```
127
128
## Usage Examples
129
130
### Working with Any Messages
131
132
```python
133
from google.api_core import protobuf_helpers
134
from google.protobuf import any_pb2
135
from my_api_pb2 import UserProfile, Settings
136
137
# Create an Any message containing a UserProfile
138
user_profile = UserProfile(name="John Doe", email="john@example.com")
139
any_message = any_pb2.Any()
140
any_message.Pack(user_profile)
141
142
# Extract the UserProfile from Any message
143
extracted_profile = protobuf_helpers.from_any_pb(UserProfile, any_message)
144
print(f"Name: {extracted_profile.name}")
145
print(f"Email: {extracted_profile.email}")
146
147
# Handle different message types
148
def process_any_message(any_msg):
149
"""Process Any message based on its type."""
150
try:
151
user = protobuf_helpers.from_any_pb(UserProfile, any_msg)
152
print(f"Processing user: {user.name}")
153
except TypeError:
154
try:
155
settings = protobuf_helpers.from_any_pb(Settings, any_msg)
156
print(f"Processing settings: {settings.theme}")
157
except TypeError:
158
print("Unknown message type")
159
```
160
161
### Field Manipulation
162
163
```python
164
from google.api_core import protobuf_helpers
165
from my_api_pb2 import UserProfile
166
167
# Create a protobuf message
168
user = UserProfile()
169
170
# Set fields using helper
171
protobuf_helpers.set(user, "name", "Alice Smith")
172
protobuf_helpers.set(user, "email", "alice@example.com")
173
protobuf_helpers.set(user, "age", 30)
174
175
# Get fields using helper
176
name = protobuf_helpers.get(user, "name")
177
email = protobuf_helpers.get(user, "email")
178
phone = protobuf_helpers.get(user, "phone", "Not provided")
179
180
print(f"Name: {name}")
181
print(f"Email: {email}")
182
print(f"Phone: {phone}")
183
184
# Use setdefault for optional fields
185
protobuf_helpers.setdefault(user, "status", "active")
186
protobuf_helpers.setdefault(user, "preferences", {})
187
188
# Works with dictionaries too
189
user_dict = {"name": "Bob", "email": "bob@example.com"}
190
protobuf_helpers.set(user_dict, "age", 25)
191
age = protobuf_helpers.get(user_dict, "age", 0)
192
```
193
194
### Field Mask Creation
195
196
```python
197
from google.api_core import protobuf_helpers
198
from my_api_pb2 import UserProfile
199
200
# Original message
201
original = UserProfile(
202
name="John Doe",
203
email="john@example.com",
204
age=30,
205
status="active"
206
)
207
208
# Modified message
209
modified = UserProfile(
210
name="John Smith", # Changed
211
email="john@example.com", # Unchanged
212
age=31, # Changed
213
status="active" # Unchanged
214
)
215
216
# Create field mask for changes
217
mask = protobuf_helpers.field_mask(original, modified)
218
print("Changed fields:", mask.paths) # ['name', 'age']
219
220
# Use field mask in API update
221
def update_user(user_id, updated_user, field_mask):
222
"""Update user with only specified fields."""
223
request = UpdateUserRequest(
224
user_id=user_id,
225
user=updated_user,
226
update_mask=field_mask
227
)
228
return client.UpdateUser(request)
229
230
response = update_user("user123", modified, mask)
231
```
232
233
### Message Type Discovery
234
235
```python
236
from google.api_core import protobuf_helpers
237
import my_api_pb2
238
239
# Discover all message types in module
240
messages = protobuf_helpers.get_messages(my_api_pb2)
241
242
print("Available message types:")
243
for name, cls in messages.items():
244
print(f" {name}: {cls}")
245
246
# Create messages dynamically
247
def create_message_by_name(message_name, **kwargs):
248
"""Create a protobuf message by name."""
249
messages = protobuf_helpers.get_messages(my_api_pb2)
250
if message_name not in messages:
251
raise ValueError(f"Unknown message type: {message_name}")
252
253
message_class = messages[message_name]
254
message = message_class()
255
256
# Set fields from kwargs
257
for key, value in kwargs.items():
258
protobuf_helpers.set(message, key, value)
259
260
return message
261
262
# Usage
263
user = create_message_by_name("UserProfile",
264
name="Jane Doe",
265
email="jane@example.com")
266
```
267
268
### Oneof Field Validation
269
270
```python
271
from google.api_core import protobuf_helpers
272
273
def create_notification(**kwargs):
274
"""Create notification with exactly one delivery method."""
275
# Validate exactly one delivery method is specified
276
delivery_method, delivery_config = protobuf_helpers.check_oneof(
277
email=kwargs.get("email"),
278
sms=kwargs.get("sms"),
279
push=kwargs.get("push")
280
)
281
282
print(f"Using {delivery_method} delivery: {delivery_config}")
283
284
# Create notification with validated delivery method
285
notification = NotificationRequest(message=kwargs["message"])
286
protobuf_helpers.set(notification, delivery_method, delivery_config)
287
288
return notification
289
290
# Valid usage - exactly one delivery method
291
notification1 = create_notification(
292
message="Hello World",
293
email="user@example.com"
294
)
295
296
# Invalid usage - multiple delivery methods (raises TypeError)
297
try:
298
notification2 = create_notification(
299
message="Hello World",
300
email="user@example.com",
301
sms="+1234567890" # This will raise TypeError
302
)
303
except TypeError as e:
304
print(f"Validation error: {e}")
305
```
306
307
### Complex Field Operations
308
309
```python
310
from google.api_core import protobuf_helpers
311
from my_api_pb2 import UserProfile, Address
312
313
def update_nested_fields():
314
"""Demonstrate working with nested protobuf fields."""
315
user = UserProfile()
316
317
# Set nested address fields
318
address = Address(
319
street="123 Main St",
320
city="Anytown",
321
state="CA",
322
zip_code="12345"
323
)
324
protobuf_helpers.set(user, "address", address)
325
326
# Get nested field values
327
street = protobuf_helpers.get(user.address, "street")
328
city = protobuf_helpers.get(user.address, "city")
329
330
print(f"Address: {street}, {city}")
331
332
# Update nested fields
333
protobuf_helpers.set(user.address, "street", "456 Oak Ave")
334
335
# Use setdefault for optional nested fields
336
protobuf_helpers.setdefault(user, "preferences", {})
337
protobuf_helpers.setdefault(user.preferences, "theme", "light")
338
protobuf_helpers.setdefault(user.preferences, "language", "en")
339
340
update_nested_fields()
341
```
342
343
### Type-Safe Field Access
344
345
```python
346
from google.api_core import protobuf_helpers
347
from typing import Optional, Any
348
349
def safe_get_field(message, field_name: str, expected_type: type, default: Any = None) -> Any:
350
"""Safely get a field with type checking."""
351
value = protobuf_helpers.get(message, field_name, default)
352
353
if value is not None and not isinstance(value, expected_type):
354
raise TypeError(f"Field {field_name} expected {expected_type}, got {type(value)}")
355
356
return value
357
358
def safe_set_field(message, field_name: str, value: Any, expected_type: type) -> None:
359
"""Safely set a field with type checking."""
360
if value is not None and not isinstance(value, expected_type):
361
raise TypeError(f"Field {field_name} expected {expected_type}, got {type(value)}")
362
363
protobuf_helpers.set(message, field_name, value)
364
365
# Usage with type safety
366
user = UserProfile()
367
368
# Type-safe field operations
369
safe_set_field(user, "name", "John Doe", str)
370
safe_set_field(user, "age", 30, int)
371
372
name = safe_get_field(user, "name", str, "Unknown")
373
age = safe_get_field(user, "age", int, 0)
374
375
print(f"User: {name}, Age: {age}")
376
```