A better Protobuf / gRPC generator & library
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Helper functions for message introspection, one-of field handling, and casing conversion to support generated code and user applications.
Function to inspect and work with protobuf one-of field groups.
def which_one_of(message: Message, group_name: str) -> Tuple[str, Any]:
"""
Return the name and value of a message's one-of field group.
Args:
message: Message instance to inspect
group_name: Name of the one-of group
Returns:
Tuple of (field_name, field_value) for the active field,
or ("", None) if no field in the group is set
"""Function to convert strings to snake_case while avoiding Python keywords.
def safe_snake_case(value: str) -> str:
"""
Snake case a value taking into account Python keywords.
Converts the input string to snake_case and appends an underscore
if the result is a Python keyword to avoid naming conflicts.
Args:
value: String to convert to snake case
Returns:
Snake-cased string, with trailing underscore if it's a Python keyword
"""Functions for working with protobuf timestamp and duration types.
def datetime_default_gen() -> datetime:
"""
Generate default datetime value for protobuf timestamps.
Returns:
datetime object representing Unix epoch (1970-01-01 UTC)
"""
# Default datetime constant
DATETIME_ZERO: datetime # 1970-01-01T00:00:00+00:00Function to check message serialization state (already covered in serialization docs but included here for completeness).
def serialized_on_wire(message: Message) -> bool:
"""
Check if this message was or should be serialized on the wire.
This can be used to detect presence (e.g. optional wrapper message)
and is used internally during parsing/serialization.
Args:
message: Message instance to check
Returns:
True if message was or should be serialized on the wire
"""from dataclasses import dataclass
import betterproto
@dataclass
class Contact(betterproto.Message):
name: str = betterproto.string_field(1)
# One-of group for contact method
email: str = betterproto.string_field(2, group="contact_method")
phone: str = betterproto.string_field(3, group="contact_method")
address: str = betterproto.string_field(4, group="contact_method")
# Create contact with email
contact = Contact(name="Alice", email="alice@example.com")
# Check which field is active
field_name, field_value = betterproto.which_one_of(contact, "contact_method")
print(f"Contact method: {field_name} = {field_value}")
# Output: Contact method: email = alice@example.com
# Switch to phone number
contact.phone = "+1-555-1234" # This clears email automatically
field_name, field_value = betterproto.which_one_of(contact, "contact_method")
print(f"Contact method: {field_name} = {field_value}")
# Output: Contact method: phone = +1-555-1234
# Check empty one-of group
empty_contact = Contact(name="Bob")
field_name, field_value = betterproto.which_one_of(empty_contact, "contact_method")
print(f"Contact method: {field_name} = {field_value}")
# Output: Contact method: = Noneimport betterproto
# Normal case conversion
result = betterproto.safe_snake_case("CamelCaseValue")
print(result) # camel_case_value
result = betterproto.safe_snake_case("XMLHttpRequest")
print(result) # xml_http_request
# Handling Python keywords
result = betterproto.safe_snake_case("class")
print(result) # class_
result = betterproto.safe_snake_case("for")
print(result) # for_
result = betterproto.safe_snake_case("import")
print(result) # import_
# Mixed cases
result = betterproto.safe_snake_case("ClassFactory")
print(result) # class_factory (not affected since "class" is in the middle)import betterproto
from datetime import datetime, timezone
# Get default datetime for protobuf timestamps
default_dt = betterproto.datetime_default_gen()
print(default_dt) # 1970-01-01 00:00:00+00:00
# Use the constant
print(betterproto.DATETIME_ZERO) # 1970-01-01 00:00:00+00:00
# Compare with current time
current = datetime.now(timezone.utc)
print(f"Seconds since epoch: {(current - betterproto.DATETIME_ZERO).total_seconds()}")from dataclasses import dataclass
@dataclass
class Person(betterproto.Message):
name: str = betterproto.string_field(1)
age: int = betterproto.int32_field(2)
@dataclass
class Group(betterproto.Message):
leader: Person = betterproto.message_field(1)
members: List[Person] = betterproto.message_field(2)
# Check serialization state
group = Group()
# Nested message starts as not serialized
print(betterproto.serialized_on_wire(group.leader)) # False
# Setting a field marks it as serialized
group.leader.name = "Alice"
print(betterproto.serialized_on_wire(group.leader)) # True
# Even setting default values marks as serialized
group.leader.age = 0 # Default value for int32
print(betterproto.serialized_on_wire(group.leader)) # Still True
# Adding to repeated field
group.members.append(Person(name="Bob"))
print(betterproto.serialized_on_wire(group.members[0])) # True# Example of how utilities work together in practice
@dataclass
class ApiRequest(betterproto.Message):
# One-of for request type
get_user: str = betterproto.string_field(1, group="request_type")
create_user: str = betterproto.string_field(2, group="request_type")
update_user: str = betterproto.string_field(3, group="request_type")
# Optional metadata
metadata: Dict[str, str] = betterproto.map_field(4, "string", "string")
def process_request(request: ApiRequest):
"""Process API request based on active one-of field."""
# Use which_one_of to determine request type
request_type, request_data = betterproto.which_one_of(request, "request_type")
if not request_type:
raise ValueError("No request type specified")
# Convert to snake_case for method dispatch
method_name = betterproto.safe_snake_case(f"handle_{request_type}")
print(f"Processing {request_type}: {request_data}")
print(f"Would call method: {method_name}")
# Check if metadata was provided
if betterproto.serialized_on_wire(request) and request.metadata:
print(f"With metadata: {request.metadata}")
# Example usage
request = ApiRequest(get_user="user123", metadata={"client": "web"})
process_request(request)
# Output:
# Processing get_user: user123
# Would call method: handle_get_user
# With metadata: {'client': 'web'}def safe_one_of_access(message: betterproto.Message, group: str):
"""Safely access one-of field with error handling."""
try:
field_name, field_value = betterproto.which_one_of(message, group)
if field_name:
return field_name, field_value
else:
return None, None
except (AttributeError, KeyError) as e:
print(f"Error accessing one-of group '{group}': {e}")
return None, None
def safe_snake_case_conversion(value: str) -> str:
"""Safely convert to snake case with validation."""
if not isinstance(value, str):
raise TypeError(f"Expected string, got {type(value)}")
if not value:
return ""
return betterproto.safe_snake_case(value)
# Usage with error handling
contact = Contact(name="Test", email="test@example.com")
# Safe one-of access
field_name, field_value = safe_one_of_access(contact, "contact_method")
if field_name:
print(f"Active field: {field_name}")
# Safe conversion
try:
snake_name = safe_snake_case_conversion("InvalidClassName")
print(f"Converted: {snake_name}")
except TypeError as e:
print(f"Conversion error: {e}")# Default datetime for protobuf timestamps
DATETIME_ZERO: datetime # datetime(1970, 1, 1, tzinfo=timezone.utc)
# Python keywords that trigger underscore suffix
PYTHON_KEYWORDS: List[str] = [
"and", "as", "assert", "break", "class", "continue", "def", "del",
"elif", "else", "except", "finally", "for", "from", "global", "if",
"import", "in", "is", "lambda", "nonlocal", "not", "or", "pass",
"raise", "return", "try", "while", "with", "yield"
]Install with Tessl CLI
npx tessl i tessl/pypi-betterproto