File size: 3,500 Bytes
1380717 |
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 |
# 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 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
|