Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/fastavro/io/binary_encoder.py: 44%
52 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:10 +0000
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:10 +0000
1from struct import pack
2from binascii import crc32
5class BinaryEncoder:
6 """Encoder for the avro binary format.
8 NOTE: All attributes and methods on this class should be considered
9 private.
11 Parameters
12 ----------
13 fo: file-like
14 Input stream
16 """
18 def __init__(self, fo):
19 self._fo = fo
21 def flush(self):
22 pass
24 def write_null(self):
25 pass
27 def write_boolean(self, datum):
28 self._fo.write(pack("B", 1 if datum else 0))
30 def write_int(self, datum):
31 datum = (datum << 1) ^ (datum >> 63)
32 while (datum & ~0x7F) != 0:
33 self._fo.write(pack("B", (datum & 0x7F) | 0x80))
34 datum >>= 7
35 self._fo.write(pack("B", datum))
37 write_long = write_int
39 def write_float(self, datum):
40 self._fo.write(pack("<f", datum))
42 def write_double(self, datum):
43 self._fo.write(pack("<d", datum))
45 def write_bytes(self, datum):
46 self.write_long(len(datum))
47 self._fo.write(datum)
49 def write_utf8(self, datum):
50 try:
51 encoded = datum.encode()
52 except AttributeError:
53 raise TypeError("must be string")
54 self.write_bytes(encoded)
56 def write_crc32(self, datum):
57 data = crc32(datum) & 0xFFFFFFFF
58 self._fo.write(pack(">I", data))
60 def write_fixed(self, datum):
61 self._fo.write(datum)
63 def write_enum(self, index):
64 self.write_int(index)
66 def write_array_start(self):
67 pass
69 def write_item_count(self, length):
70 self.write_long(length)
72 def end_item(self):
73 pass
75 def write_array_end(self):
76 self.write_long(0)
78 def write_map_start(self):
79 pass
81 def write_map_end(self):
82 self.write_long(0)
84 def write_index(self, index, schema=None):
85 self.write_long(index)