0
# Utility Functions
1
2
Helper functions for message introspection, one-of field handling, and casing conversion to support generated code and user applications.
3
4
## Capabilities
5
6
### One-of Field Handling
7
8
Function to inspect and work with protobuf one-of field groups.
9
10
```python { .api }
11
def which_one_of(message: Message, group_name: str) -> Tuple[str, Any]:
12
"""
13
Return the name and value of a message's one-of field group.
14
15
Args:
16
message: Message instance to inspect
17
group_name: Name of the one-of group
18
19
Returns:
20
Tuple of (field_name, field_value) for the active field,
21
or ("", None) if no field in the group is set
22
"""
23
```
24
25
### String Case Conversion
26
27
Function to convert strings to snake_case while avoiding Python keywords.
28
29
```python { .api }
30
def safe_snake_case(value: str) -> str:
31
"""
32
Snake case a value taking into account Python keywords.
33
34
Converts the input string to snake_case and appends an underscore
35
if the result is a Python keyword to avoid naming conflicts.
36
37
Args:
38
value: String to convert to snake case
39
40
Returns:
41
Snake-cased string, with trailing underscore if it's a Python keyword
42
"""
43
```
44
45
### DateTime Utilities
46
47
Functions for working with protobuf timestamp and duration types.
48
49
```python { .api }
50
def datetime_default_gen() -> datetime:
51
"""
52
Generate default datetime value for protobuf timestamps.
53
54
Returns:
55
datetime object representing Unix epoch (1970-01-01 UTC)
56
"""
57
58
# Default datetime constant
59
DATETIME_ZERO: datetime # 1970-01-01T00:00:00+00:00
60
```
61
62
### Message State Inspection
63
64
Function to check message serialization state (already covered in serialization docs but included here for completeness).
65
66
```python { .api }
67
def serialized_on_wire(message: Message) -> bool:
68
"""
69
Check if this message was or should be serialized on the wire.
70
71
This can be used to detect presence (e.g. optional wrapper message)
72
and is used internally during parsing/serialization.
73
74
Args:
75
message: Message instance to check
76
77
Returns:
78
True if message was or should be serialized on the wire
79
"""
80
```
81
82
## Usage Examples
83
84
### Working with One-of Fields
85
86
```python
87
from dataclasses import dataclass
88
import betterproto
89
90
@dataclass
91
class Contact(betterproto.Message):
92
name: str = betterproto.string_field(1)
93
94
# One-of group for contact method
95
email: str = betterproto.string_field(2, group="contact_method")
96
phone: str = betterproto.string_field(3, group="contact_method")
97
address: str = betterproto.string_field(4, group="contact_method")
98
99
# Create contact with email
100
contact = Contact(name="Alice", email="alice@example.com")
101
102
# Check which field is active
103
field_name, field_value = betterproto.which_one_of(contact, "contact_method")
104
print(f"Contact method: {field_name} = {field_value}")
105
# Output: Contact method: email = alice@example.com
106
107
# Switch to phone number
108
contact.phone = "+1-555-1234" # This clears email automatically
109
field_name, field_value = betterproto.which_one_of(contact, "contact_method")
110
print(f"Contact method: {field_name} = {field_value}")
111
# Output: Contact method: phone = +1-555-1234
112
113
# Check empty one-of group
114
empty_contact = Contact(name="Bob")
115
field_name, field_value = betterproto.which_one_of(empty_contact, "contact_method")
116
print(f"Contact method: {field_name} = {field_value}")
117
# Output: Contact method: = None
118
```
119
120
### Safe Snake Case Conversion
121
122
```python
123
import betterproto
124
125
# Normal case conversion
126
result = betterproto.safe_snake_case("CamelCaseValue")
127
print(result) # camel_case_value
128
129
result = betterproto.safe_snake_case("XMLHttpRequest")
130
print(result) # xml_http_request
131
132
# Handling Python keywords
133
result = betterproto.safe_snake_case("class")
134
print(result) # class_
135
136
result = betterproto.safe_snake_case("for")
137
print(result) # for_
138
139
result = betterproto.safe_snake_case("import")
140
print(result) # import_
141
142
# Mixed cases
143
result = betterproto.safe_snake_case("ClassFactory")
144
print(result) # class_factory (not affected since "class" is in the middle)
145
```
146
147
### DateTime Default Generation
148
149
```python
150
import betterproto
151
from datetime import datetime, timezone
152
153
# Get default datetime for protobuf timestamps
154
default_dt = betterproto.datetime_default_gen()
155
print(default_dt) # 1970-01-01 00:00:00+00:00
156
157
# Use the constant
158
print(betterproto.DATETIME_ZERO) # 1970-01-01 00:00:00+00:00
159
160
# Compare with current time
161
current = datetime.now(timezone.utc)
162
print(f"Seconds since epoch: {(current - betterproto.DATETIME_ZERO).total_seconds()}")
163
```
164
165
### Message State Checking
166
167
```python
168
from dataclasses import dataclass
169
170
@dataclass
171
class Person(betterproto.Message):
172
name: str = betterproto.string_field(1)
173
age: int = betterproto.int32_field(2)
174
175
@dataclass
176
class Group(betterproto.Message):
177
leader: Person = betterproto.message_field(1)
178
members: List[Person] = betterproto.message_field(2)
179
180
# Check serialization state
181
group = Group()
182
183
# Nested message starts as not serialized
184
print(betterproto.serialized_on_wire(group.leader)) # False
185
186
# Setting a field marks it as serialized
187
group.leader.name = "Alice"
188
print(betterproto.serialized_on_wire(group.leader)) # True
189
190
# Even setting default values marks as serialized
191
group.leader.age = 0 # Default value for int32
192
print(betterproto.serialized_on_wire(group.leader)) # Still True
193
194
# Adding to repeated field
195
group.members.append(Person(name="Bob"))
196
print(betterproto.serialized_on_wire(group.members[0])) # True
197
```
198
199
### Combining Utilities in Generated Code
200
201
```python
202
# Example of how utilities work together in practice
203
@dataclass
204
class ApiRequest(betterproto.Message):
205
# One-of for request type
206
get_user: str = betterproto.string_field(1, group="request_type")
207
create_user: str = betterproto.string_field(2, group="request_type")
208
update_user: str = betterproto.string_field(3, group="request_type")
209
210
# Optional metadata
211
metadata: Dict[str, str] = betterproto.map_field(4, "string", "string")
212
213
def process_request(request: ApiRequest):
214
"""Process API request based on active one-of field."""
215
216
# Use which_one_of to determine request type
217
request_type, request_data = betterproto.which_one_of(request, "request_type")
218
219
if not request_type:
220
raise ValueError("No request type specified")
221
222
# Convert to snake_case for method dispatch
223
method_name = betterproto.safe_snake_case(f"handle_{request_type}")
224
225
print(f"Processing {request_type}: {request_data}")
226
print(f"Would call method: {method_name}")
227
228
# Check if metadata was provided
229
if betterproto.serialized_on_wire(request) and request.metadata:
230
print(f"With metadata: {request.metadata}")
231
232
# Example usage
233
request = ApiRequest(get_user="user123", metadata={"client": "web"})
234
process_request(request)
235
# Output:
236
# Processing get_user: user123
237
# Would call method: handle_get_user
238
# With metadata: {'client': 'web'}
239
```
240
241
### Error Handling with Utilities
242
243
```python
244
def safe_one_of_access(message: betterproto.Message, group: str):
245
"""Safely access one-of field with error handling."""
246
try:
247
field_name, field_value = betterproto.which_one_of(message, group)
248
if field_name:
249
return field_name, field_value
250
else:
251
return None, None
252
except (AttributeError, KeyError) as e:
253
print(f"Error accessing one-of group '{group}': {e}")
254
return None, None
255
256
def safe_snake_case_conversion(value: str) -> str:
257
"""Safely convert to snake case with validation."""
258
if not isinstance(value, str):
259
raise TypeError(f"Expected string, got {type(value)}")
260
261
if not value:
262
return ""
263
264
return betterproto.safe_snake_case(value)
265
266
# Usage with error handling
267
contact = Contact(name="Test", email="test@example.com")
268
269
# Safe one-of access
270
field_name, field_value = safe_one_of_access(contact, "contact_method")
271
if field_name:
272
print(f"Active field: {field_name}")
273
274
# Safe conversion
275
try:
276
snake_name = safe_snake_case_conversion("InvalidClassName")
277
print(f"Converted: {snake_name}")
278
except TypeError as e:
279
print(f"Conversion error: {e}")
280
```
281
282
## Constants
283
284
```python { .api }
285
# Default datetime for protobuf timestamps
286
DATETIME_ZERO: datetime # datetime(1970, 1, 1, tzinfo=timezone.utc)
287
288
# Python keywords that trigger underscore suffix
289
PYTHON_KEYWORDS: List[str] = [
290
"and", "as", "assert", "break", "class", "continue", "def", "del",
291
"elif", "else", "except", "finally", "for", "from", "global", "if",
292
"import", "in", "is", "lambda", "nonlocal", "not", "or", "pass",
293
"raise", "return", "try", "while", "with", "yield"
294
]
295
```