Marshal System

class Marshal:
    """
    The translator between protocol buffer and Python instances.
    
    Multiple instantiations with the same name provide the same instance,
    enabling shared marshal registries across modules.
    
    Args:
        name (str): The name of the marshal. Multiple marshals with the same
                   name argument will provide the same marshal instance.
    """
    def __init__(self, *, name: str): ...
    
    def register(self, proto_type: type, rule: Rule = None): ...
    def to_python(self, proto_type, value, *, absent: bool = None): ...
    def to_proto(self, proto_type, value, *, strict: bool = False): ...
    def reset(self): ...

# Create a marshal instance
marshal = proto.Marshal(name="my_app")

# Multiple references to the same marshal
marshal2 = proto.Marshal(name="my_app")  # Same instance as marshal

# Each proto-plus message automatically gets its own marshal
# based on the module/package structure

def register(self, proto_type: type, rule: Rule = None):
    """
    Register a rule against the given proto_type.
    
    Args:
        proto_type (type): A protocol buffer message type.
        rule: A marshal object with to_python and to_proto methods.
        
    Can also be used as a decorator for rule classes.
    """

from google.protobuf import timestamp_pb2
import datetime

# Create a custom rule class
class CustomTimestampRule:
    def to_python(self, pb_value, *, absent: bool = None):
        if absent:
            return None
        return datetime.datetime.fromtimestamp(
            pb_value.seconds + pb_value.nanos / 1e9
        )
    
    def to_proto(self, py_value):
        if py_value is None:
            return timestamp_pb2.Timestamp()
        timestamp = int(py_value.timestamp())
        return timestamp_pb2.Timestamp(seconds=timestamp)

# Register the rule
marshal = proto.Marshal(name="custom")
marshal.register(timestamp_pb2.Timestamp, CustomTimestampRule())

# Or use as a decorator
@marshal.register(timestamp_pb2.Timestamp)
class TimestampRule:
    def to_python(self, pb_value, *, absent: bool = None):
        # Custom conversion logic
        pass
    
    def to_proto(self, py_value):
        # Custom conversion logic  
        pass

def to_python(self, proto_type, value, *, absent: bool = None):
    """
    Convert a protocol buffer value to the appropriate Python type.
    
    Args:
        proto_type: The protocol buffer type descriptor
        value: The protocol buffer value to convert
        absent (bool): Whether the field was absent in the message
        
    Returns:
        The converted Python value
    """

from google.protobuf import timestamp_pb2
import datetime

marshal = proto.Marshal(name="example")

# Convert timestamp to Python datetime
pb_timestamp = timestamp_pb2.Timestamp()
pb_timestamp.GetCurrentTime()

py_datetime = marshal.to_python(timestamp_pb2.Timestamp, pb_timestamp)
print(type(py_datetime))  # <class 'datetime.datetime'>

# Convert repeated fields to Python lists
# (handled automatically by the marshal)

def to_proto(self, proto_type, value, *, strict: bool = False):
    """
    Convert a Python value to the appropriate protocol buffer type.
    
    Args:
        proto_type: The target protocol buffer type
        value: The Python value to convert
        strict (bool): If True, require exact type match
        
    Returns:
        The converted protocol buffer value
    """

from google.protobuf import timestamp_pb2
import datetime

marshal = proto.Marshal(name="example")

# Convert Python datetime to timestamp
py_datetime = datetime.datetime.now()
pb_timestamp = marshal.to_proto(timestamp_pb2.Timestamp, py_datetime)
print(type(pb_timestamp))  # <class 'google.protobuf.timestamp_pb2.Timestamp'>

# Strict mode validation
try:
    result = marshal.to_proto(timestamp_pb2.Timestamp, "invalid", strict=True)
except TypeError as e:
    print(f"Type mismatch: {e}")

def reset(self):
    """Reset the registry to its initial state."""

marshal = proto.Marshal(name="example")

# Add custom rules
marshal.register(SomeType, CustomRule())

# Reset to default rules only
marshal.reset()

from google.protobuf import timestamp_pb2, duration_pb2, wrappers_pb2
import datetime

class Event(proto.Message):
    # Timestamp converts to/from datetime.datetime
    created_at = proto.Field(timestamp_pb2.Timestamp, number=1)
    
    # Duration converts to/from datetime.timedelta  
    duration = proto.Field(duration_pb2.Duration, number=2)
    
    # Wrapper types convert to/from native Python types
    description = proto.Field(wrappers_pb2.StringValue, number=3)
    count = proto.Field(wrappers_pb2.Int32Value, number=4)

# Usage with automatic conversion
event = Event(
    created_at=datetime.datetime.now(),           # datetime -> Timestamp
    duration=datetime.timedelta(hours=2),         # timedelta -> Duration  
    description="Sample event",                   # str -> StringValue
    count=42                                      # int -> Int32Value
)

# Access converted values
print(type(event.created_at))    # <class 'datetime.datetime'>
print(type(event.duration))      # <class 'datetime.timedelta'>
print(type(event.description))   # <class 'str'>
print(type(event.count))         # <class 'int'>

from google.protobuf import struct_pb2

class ApiResponse(proto.Message):
    # Struct converts to/from Python dict
    metadata = proto.Field(struct_pb2.Struct, number=1)
    
    # Value converts to/from Python native types
    result = proto.Field(struct_pb2.Value, number=2)

# Usage with automatic conversion
response = ApiResponse(
    metadata={
        "request_id": "12345",
        "timestamp": 1640995200,
        "success": True
    },
    result=[1, 2, 3, "test"]  # Converted to appropriate Value type
)

# Access as native Python types
print(response.metadata["success"])  # True (bool)
print(response.result[3])            # "test" (str)

from google.protobuf import field_mask_pb2

class UpdateRequest(proto.Message):
    # FieldMask converts to/from list of field paths
    update_mask = proto.Field(field_mask_pb2.FieldMask, number=1)

# Usage
request = UpdateRequest(
    update_mask=["name", "email", "address.city"]  # List -> FieldMask
)

# Access as list
for path in request.update_mask:
    print(f"Update path: {path}")

class DataSet(proto.Message):
    # Repeated fields become Python lists
    values = proto.RepeatedField(proto.FLOAT, number=1)
    
    # Map fields become Python dictionaries  
    labels = proto.MapField(proto.STRING, proto.STRING, number=2)

data = DataSet(
    values=[1.0, 2.5, 3.7],
    labels={"category": "test", "version": "1.0"}
)

# Collections behave like native Python types
data.values.append(4.2)
data.labels["new_key"] = "new_value"

print(len(data.values))      # 4
print(list(data.labels.keys()))  # ["category", "version", "new_key"]

class FileData(proto.Message):
    filename = proto.Field(proto.STRING, number=1)
    content = proto.Field(proto.BYTES, number=2)

# Bytes fields accept bytes or base64 strings
file_data = FileData(
    filename="example.txt",
    content=b"Hello, World!"  # Raw bytes
)

# Can also use base64 strings in JSON
json_data = '{"filename": "test.txt", "content": "SGVsbG8sIFdvcmxkIQ=="}'
file_from_json = FileData.from_json(json_data)
print(file_from_json.content)  # b'Hello, World!'

from proto.marshal import Rule

class ColorRule(Rule):
    """Convert between RGB tuples and color messages."""
    
    def to_python(self, pb_value, *, absent: bool = None):
        if absent:
            return None
        return (pb_value.red, pb_value.green, pb_value.blue)
    
    def to_proto(self, py_value):
        if isinstance(py_value, tuple) and len(py_value) == 3:
            return ColorMessage(red=py_value[0], green=py_value[1], blue=py_value[2])
        return py_value

# Register and use the custom rule
marshal = proto.Marshal(name="graphics")
marshal.register(ColorMessage, ColorRule())