Skip to content

BinaryWriter

faststream.redis.parser.binary.BinaryWriter #

BinaryWriter()
Source code in faststream/redis/parser/binary.py 
101
102
def __init__(self) -> None:
    self.data = bytearray()

data instance-attribute #

data = bytearray()

write #

write(data)
Source code in faststream/redis/parser/binary.py 
104
105
def write(self, data: bytes) -> None:
    self.data.extend(data)

write_short #

write_short(number)
Source code in faststream/redis/parser/binary.py 
107
108
109
def write_short(self, number: int) -> None:
    int_bytes = pack(">H", number)
    self.write(int_bytes)

write_int #

write_int(number)
Source code in faststream/redis/parser/binary.py 
111
112
113
def write_int(self, number: int) -> None:
    int_bytes = pack(">I", number)
    self.write(int_bytes)

write_string #

write_string(data)
Source code in faststream/redis/parser/binary.py 
115
116
117
118
119
120
121
def write_string(self, data: str | bytes) -> None:
    str_len = len(data)
    self.write_short(str_len)
    if isinstance(data, bytes):
        self.write(data)
    else:
        self.write(data.encode())

get_bytes #

get_bytes()
Source code in faststream/redis/parser/binary.py 
123
124
def get_bytes(self) -> bytes:
    return bytes(self.data)