File size: 4,389 Bytes
9c6594c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 |
# Protocol Buffers - Google's data interchange format
# Copyright 2008 Google Inc. All rights reserved.
#
# Use of this source code is governed by a BSD-style
# license that can be found in the LICENSE file or at
# https://developers.google.com/open-source/licenses/bsd
"""Contains the Nextgen Pythonic protobuf APIs."""
import io
from typing import Text, Type, TypeVar
from google.protobuf.internal import decoder
from google.protobuf.internal import encoder
from google.protobuf.message import Message
_MESSAGE = TypeVar('_MESSAGE', bound='Message')
def serialize(message: _MESSAGE, deterministic: bool = None) -> bytes:
"""Return the serialized proto.
Args:
message: The proto message to be serialized.
deterministic: If true, requests deterministic serialization
of the protobuf, with predictable ordering of map keys.
Returns:
A binary bytes representation of the message.
"""
return message.SerializeToString(deterministic=deterministic)
def parse(message_class: Type[_MESSAGE], payload: bytes) -> _MESSAGE:
"""Given a serialized data in binary form, deserialize it into a Message.
Args:
message_class: The message meta class.
payload: A serialized bytes in binary form.
Returns:
A new message deserialized from payload.
"""
new_message = message_class()
new_message.ParseFromString(payload)
return new_message
def serialize_length_prefixed(message: _MESSAGE, output: io.BytesIO) -> None:
"""Writes the size of the message as a varint and the serialized message.
Writes the size of the message as a varint and then the serialized message.
This allows more data to be written to the output after the message. Use
parse_length_prefixed to parse messages written by this method.
The output stream must be buffered, e.g. using
https://docs.python.org/3/library/io.html#buffered-streams.
Example usage:
out = io.BytesIO()
for msg in message_list:
proto.serialize_length_prefixed(msg, out)
Args:
message: The protocol buffer message that should be serialized.
output: BytesIO or custom buffered IO that data should be written to.
"""
size = message.ByteSize()
encoder._VarintEncoder()(output.write, size)
out_size = output.write(serialize(message))
if out_size != size:
raise TypeError(
'Failed to write complete message (wrote: %d, expected: %d)'
'. Ensure output is using buffered IO.' % (out_size, size)
)
def parse_length_prefixed(
message_class: Type[_MESSAGE], input_bytes: io.BytesIO
) -> _MESSAGE:
"""Parse a message from input_bytes.
Args:
message_class: The protocol buffer message class that parser should parse.
input_bytes: A buffered input.
Example usage:
while True:
msg = proto.parse_length_prefixed(message_class, input_bytes)
if msg is None:
break
...
Returns:
A parsed message if successful. None if input_bytes is at EOF.
"""
size = decoder._DecodeVarint(input_bytes)
if size is None:
# It is the end of buffered input. See example usage in the
# API description.
return None
message = message_class()
if size == 0:
return message
parsed_size = message.ParseFromString(input_bytes.read(size))
if parsed_size != size:
raise ValueError(
'Truncated message or non-buffered input_bytes: '
'Expected {0} bytes but only {1} bytes parsed for '
'{2}.'.format(size, parsed_size, message.DESCRIPTOR.name)
)
return message
def byte_size(message: Message) -> int:
"""Returns the serialized size of this message.
Args:
message: A proto message.
Returns:
int: The number of bytes required to serialize this message.
"""
return message.ByteSize()
def clear_message(message: Message) -> None:
"""Clears all data that was set in the message.
Args:
message: The proto message to be cleared.
"""
message.Clear()
def clear_field(message: Message, field_name: Text) -> None:
"""Clears the contents of a given field.
Inside a oneof group, clears the field set. If the name neither refers to a
defined field or oneof group, :exc:`ValueError` is raised.
Args:
message: The proto message.
field_name (str): The name of the field to be cleared.
Raises:
ValueError: if the `field_name` is not a member of this message.
"""
message.ClearField(field_name)
|