Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/fastavro/io/binary_encoder.py: 44%

1from struct import pack 

2from binascii import crc32 

3 

4 

5class BinaryEncoder: 

6 """Encoder for the avro binary format. 

7 

8 NOTE: All attributes and methods on this class should be considered 

9 private. 

10 

11 Parameters 

12 ---------- 

13 fo: file-like 

14 Input stream 

15 

16 """ 

17 

18 def __init__(self, fo): 

19 self._fo = fo 

20 

21 def flush(self): 

22 pass 

23 

24 def write_null(self): 

25 pass 

26 

27 def write_boolean(self, datum): 

28 self._fo.write(pack("B", 1 if datum else 0)) 

29 

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)) 

36 

37 write_long = write_int 

38 

39 def write_float(self, datum): 

40 self._fo.write(pack("<f", datum)) 

41 

42 def write_double(self, datum): 

43 self._fo.write(pack("<d", datum)) 

44 

45 def write_bytes(self, datum): 

46 self.write_long(len(datum)) 

47 self._fo.write(datum) 

48 

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) 

55 

56 def write_crc32(self, datum): 

57 data = crc32(datum) & 0xFFFFFFFF 

58 self._fo.write(pack(">I", data)) 

59 

60 def write_fixed(self, datum): 

61 self._fo.write(datum) 

62 

63 def write_enum(self, index): 

64 self.write_int(index) 

65 

66 def write_array_start(self): 

67 pass 

68 

69 def write_item_count(self, length): 

70 self.write_long(length) 

71 

72 def end_item(self): 

73 pass 

74 

75 def write_array_end(self): 

76 self.write_long(0) 

77 

78 def write_map_start(self): 

79 pass 

80 

81 def write_map_end(self): 

82 self.write_long(0) 

83 

84 def write_index(self, index, schema=None): 

85 self.write_long(index)