Protocol Buffer Helpers

def from_any_pb(pb_type, any_pb):
    """
    Converts an Any protobuf message to the specified message type.
    
    Args:
        pb_type (type): Target protobuf message class
        any_pb: google.protobuf.any_pb2.Any message
        
    Returns:
        Message: Deserialized message of specified type
        
    Raises:
        TypeError: If any_pb is not compatible with pb_type
    """

def get_messages(module):
    """
    Discover all protobuf Message classes in a module.
    
    Args:
        module: Python module containing protobuf classes
        
    Returns:
        dict: Mapping of message names to message classes
    """

def check_oneof(**kwargs):
    """
    Validate that exactly one of the keyword arguments is set.
    
    Args:
        **kwargs: Keyword arguments to validate
        
    Raises:
        TypeError: If zero or more than one argument is provided
        
    Returns:
        tuple: (key, value) of the single provided argument
    """

def field_mask(original, modified):
    """
    Create a field mask by comparing two protobuf messages.
    
    Args:
        original: Original protobuf message
        modified: Modified protobuf message
        
    Returns:
        google.protobuf.field_mask_pb2.FieldMask: Field mask representing changes
    """

def get(msg_or_dict, key, default=None):
    """
    Get a value from a protobuf Message or dictionary.
    
    Args:
        msg_or_dict: Protobuf message or dictionary
        key (str): Field name or dictionary key
        default: Default value if key not found
        
    Returns:
        Any: Field value or default
    """

def set(msg_or_dict, key, value):
    """
    Set a value on a protobuf Message or dictionary.
    
    Args:
        msg_or_dict: Protobuf message or dictionary
        key (str): Field name or dictionary key  
        value: Value to set
    """

def setdefault(msg_or_dict, key, value):
    """
    Set a value only if the current value is falsy.
    
    Args:
        msg_or_dict: Protobuf message or dictionary
        key (str): Field name or dictionary key
        value: Value to set if current value is falsy
        
    Returns:
        Any: Current value (existing or newly set)
    """

# Tuple of protobuf wrapper types
_WRAPPER_TYPES = (
    "google.protobuf.wrappers_pb2.BoolValue",
    "google.protobuf.wrappers_pb2.BytesValue", 
    "google.protobuf.wrappers_pb2.DoubleValue",
    "google.protobuf.wrappers_pb2.FloatValue",
    "google.protobuf.wrappers_pb2.Int32Value",
    "google.protobuf.wrappers_pb2.Int64Value",
    "google.protobuf.wrappers_pb2.StringValue",
    "google.protobuf.wrappers_pb2.UInt32Value",
    "google.protobuf.wrappers_pb2.UInt64Value"
)

# Sentinel object for default values
_SENTINEL = object()

from google.api_core import protobuf_helpers
from google.protobuf import any_pb2
from my_api_pb2 import UserProfile, Settings

# Create an Any message containing a UserProfile
user_profile = UserProfile(name="John Doe", email="john@example.com")
any_message = any_pb2.Any()
any_message.Pack(user_profile)

# Extract the UserProfile from Any message  
extracted_profile = protobuf_helpers.from_any_pb(UserProfile, any_message)
print(f"Name: {extracted_profile.name}")
print(f"Email: {extracted_profile.email}")

# Handle different message types
def process_any_message(any_msg):
    """Process Any message based on its type."""
    try:
        user = protobuf_helpers.from_any_pb(UserProfile, any_msg)
        print(f"Processing user: {user.name}")
    except TypeError:
        try:
            settings = protobuf_helpers.from_any_pb(Settings, any_msg)
            print(f"Processing settings: {settings.theme}")
        except TypeError:
            print("Unknown message type")

from google.api_core import protobuf_helpers
from my_api_pb2 import UserProfile

# Create a protobuf message
user = UserProfile()

# Set fields using helper
protobuf_helpers.set(user, "name", "Alice Smith")
protobuf_helpers.set(user, "email", "alice@example.com")
protobuf_helpers.set(user, "age", 30)

# Get fields using helper
name = protobuf_helpers.get(user, "name")
email = protobuf_helpers.get(user, "email")
phone = protobuf_helpers.get(user, "phone", "Not provided")

print(f"Name: {name}")
print(f"Email: {email}")
print(f"Phone: {phone}")

# Use setdefault for optional fields
protobuf_helpers.setdefault(user, "status", "active")
protobuf_helpers.setdefault(user, "preferences", {})

# Works with dictionaries too
user_dict = {"name": "Bob", "email": "bob@example.com"}
protobuf_helpers.set(user_dict, "age", 25)
age = protobuf_helpers.get(user_dict, "age", 0)

from google.api_core import protobuf_helpers
from my_api_pb2 import UserProfile

# Original message
original = UserProfile(
    name="John Doe",
    email="john@example.com", 
    age=30,
    status="active"
)

# Modified message
modified = UserProfile(
    name="John Smith",  # Changed
    email="john@example.com",  # Unchanged
    age=31,  # Changed
    status="active"  # Unchanged
)

# Create field mask for changes
mask = protobuf_helpers.field_mask(original, modified)
print("Changed fields:", mask.paths)  # ['name', 'age']

# Use field mask in API update
def update_user(user_id, updated_user, field_mask):
    """Update user with only specified fields."""
    request = UpdateUserRequest(
        user_id=user_id,
        user=updated_user,
        update_mask=field_mask
    )
    return client.UpdateUser(request)

response = update_user("user123", modified, mask)

from google.api_core import protobuf_helpers
import my_api_pb2

# Discover all message types in module
messages = protobuf_helpers.get_messages(my_api_pb2)

print("Available message types:")
for name, cls in messages.items():
    print(f"  {name}: {cls}")

# Create messages dynamically
def create_message_by_name(message_name, **kwargs):
    """Create a protobuf message by name."""
    messages = protobuf_helpers.get_messages(my_api_pb2)
    if message_name not in messages:
        raise ValueError(f"Unknown message type: {message_name}")
    
    message_class = messages[message_name]
    message = message_class()
    
    # Set fields from kwargs
    for key, value in kwargs.items():
        protobuf_helpers.set(message, key, value)
    
    return message

# Usage
user = create_message_by_name("UserProfile", 
                             name="Jane Doe",
                             email="jane@example.com")

from google.api_core import protobuf_helpers

def create_notification(**kwargs):
    """Create notification with exactly one delivery method."""
    # Validate exactly one delivery method is specified
    delivery_method, delivery_config = protobuf_helpers.check_oneof(
        email=kwargs.get("email"),
        sms=kwargs.get("sms"), 
        push=kwargs.get("push")
    )
    
    print(f"Using {delivery_method} delivery: {delivery_config}")
    
    # Create notification with validated delivery method
    notification = NotificationRequest(message=kwargs["message"])
    protobuf_helpers.set(notification, delivery_method, delivery_config)
    
    return notification

# Valid usage - exactly one delivery method
notification1 = create_notification(
    message="Hello World",
    email="user@example.com"
)

# Invalid usage - multiple delivery methods (raises TypeError)
try:
    notification2 = create_notification(
        message="Hello World",
        email="user@example.com",
        sms="+1234567890"  # This will raise TypeError
    )
except TypeError as e:
    print(f"Validation error: {e}")

from google.api_core import protobuf_helpers
from my_api_pb2 import UserProfile, Address

def update_nested_fields():
    """Demonstrate working with nested protobuf fields."""
    user = UserProfile()
    
    # Set nested address fields
    address = Address(
        street="123 Main St",
        city="Anytown", 
        state="CA",
        zip_code="12345"
    )
    protobuf_helpers.set(user, "address", address)
    
    # Get nested field values
    street = protobuf_helpers.get(user.address, "street")
    city = protobuf_helpers.get(user.address, "city")
    
    print(f"Address: {street}, {city}")
    
    # Update nested fields
    protobuf_helpers.set(user.address, "street", "456 Oak Ave")
    
    # Use setdefault for optional nested fields
    protobuf_helpers.setdefault(user, "preferences", {})
    protobuf_helpers.setdefault(user.preferences, "theme", "light")
    protobuf_helpers.setdefault(user.preferences, "language", "en")

update_nested_fields()

from google.api_core import protobuf_helpers
from typing import Optional, Any

def safe_get_field(message, field_name: str, expected_type: type, default: Any = None) -> Any:
    """Safely get a field with type checking."""
    value = protobuf_helpers.get(message, field_name, default)
    
    if value is not None and not isinstance(value, expected_type):
        raise TypeError(f"Field {field_name} expected {expected_type}, got {type(value)}")
    
    return value

def safe_set_field(message, field_name: str, value: Any, expected_type: type) -> None:
    """Safely set a field with type checking."""
    if value is not None and not isinstance(value, expected_type):
        raise TypeError(f"Field {field_name} expected {expected_type}, got {type(value)}")
    
    protobuf_helpers.set(message, field_name, value)

# Usage with type safety
user = UserProfile()

# Type-safe field operations
safe_set_field(user, "name", "John Doe", str)
safe_set_field(user, "age", 30, int)

name = safe_get_field(user, "name", str, "Unknown")
age = safe_get_field(user, "age", int, 0)

print(f"User: {name}, Age: {age}")