2021-03-23 12:30:03 +00:00
|
|
|
from typing import *
|
2023-06-26 12:12:22 +00:00
|
|
|
from typing_extensions import Self
|
|
|
|
# XXX
|
|
|
|
# Note that MessageType "subclasses" are not true subclasses, but instead instances
|
|
|
|
# of the built-in metaclass MsgDef. MessageType instances are in fact instances of
|
|
|
|
# the built-in type Msg. That is why isinstance checks do not work, and instead the
|
|
|
|
# MessageTypeSubclass.is_type_of() method must be used.
|
|
|
|
|
|
|
|
|
|
|
|
# rust/src/protobuf/obj.rs
|
|
|
|
class MessageType:
|
|
|
|
MESSAGE_NAME: ClassVar[str] = "MessageType"
|
|
|
|
MESSAGE_WIRE_TYPE: ClassVar[int | None] = None
|
|
|
|
@classmethod
|
|
|
|
def is_type_of(cls: type[Self], msg: "MessageType") -> TypeGuard[Self]:
|
|
|
|
"""Identify if the provided message belongs to this type."""
|
2021-03-23 12:30:03 +00:00
|
|
|
T = TypeVar("T", bound=MessageType)
|
|
|
|
|
|
|
|
|
2023-06-26 12:12:22 +00:00
|
|
|
# rust/src/protobuf/obj.rs
|
|
|
|
def type_for_name(name: str) -> type[MessageType]:
|
2021-03-23 12:30:03 +00:00
|
|
|
"""Find the message definition for the given protobuf name."""
|
|
|
|
|
|
|
|
|
2023-06-26 12:12:22 +00:00
|
|
|
# rust/src/protobuf/obj.rs
|
|
|
|
def type_for_wire(wire_id: int) -> type[MessageType]:
|
|
|
|
"""Find the message definition for the given wire type (numeric identifier)."""
|
2021-03-23 12:30:03 +00:00
|
|
|
|
|
|
|
|
2023-06-26 12:12:22 +00:00
|
|
|
# rust/src/protobuf/obj.rs
|
2021-03-23 12:30:03 +00:00
|
|
|
def decode(
|
|
|
|
buffer: bytes,
|
2023-06-26 12:12:22 +00:00
|
|
|
msg_type: type[T],
|
2021-03-23 12:30:03 +00:00
|
|
|
enable_experimental: bool,
|
|
|
|
) -> T:
|
|
|
|
"""Decode data in the buffer into the specified message type."""
|
|
|
|
|
|
|
|
|
2023-06-26 12:12:22 +00:00
|
|
|
# rust/src/protobuf/obj.rs
|
2021-03-23 12:30:03 +00:00
|
|
|
def encoded_length(msg: MessageType) -> int:
|
|
|
|
"""Calculate length of encoding of the specified message."""
|
|
|
|
|
|
|
|
|
2023-06-26 12:12:22 +00:00
|
|
|
# rust/src/protobuf/obj.rs
|
2021-03-23 12:30:03 +00:00
|
|
|
def encode(buffer: bytearray, msg: MessageType) -> int:
|
|
|
|
"""Encode the message into the specified buffer. Return length of
|
|
|
|
encoding."""
|