Encode the value passed in and append it to the pieces list returning the the size of the encoded value. :param list pieces: Already encoded values :param any value: The value to encode :rtype: int
(pieces: List[bytes], value: Any)
| 79 | |
| 80 | |
| 81 | def encode_value(pieces: List[bytes], value: Any) -> int: # pylint: disable=R0911 |
| 82 | """Encode the value passed in and append it to the pieces list returning |
| 83 | the the size of the encoded value. |
| 84 | |
| 85 | :param list pieces: Already encoded values |
| 86 | :param any value: The value to encode |
| 87 | :rtype: int |
| 88 | |
| 89 | """ |
| 90 | |
| 91 | if isinstance(value, str): |
| 92 | value = as_bytes(value) |
| 93 | pieces.append(struct.pack('>cI', b'S', len(value))) |
| 94 | pieces.append(value) |
| 95 | return 5 + len(value) |
| 96 | elif isinstance(value, bytes): |
| 97 | pieces.append(struct.pack('>cI', b'x', len(value))) |
| 98 | pieces.append(value) |
| 99 | return 5 + len(value) |
| 100 | elif isinstance(value, bool): |
| 101 | pieces.append(struct.pack('>cB', b't', int(value))) |
| 102 | return 2 |
| 103 | elif isinstance(value, long): |
| 104 | pieces.append(struct.pack('>cq', b'l', value)) |
| 105 | return 9 |
| 106 | elif isinstance(value, int): |
| 107 | try: |
| 108 | packed = struct.pack('>ci', b'I', value) |
| 109 | pieces.append(packed) |
| 110 | return 5 |
| 111 | except struct.error: |
| 112 | pieces.append(struct.pack('>cq', b'l', long(value))) |
| 113 | return 9 |
| 114 | elif isinstance(value, decimal.Decimal): |
| 115 | value = value.normalize() |
| 116 | if value.as_tuple().exponent < 0: |
| 117 | decimals = -value.as_tuple().exponent |
| 118 | raw = int(value * (decimal.Decimal(10)**decimals)) |
| 119 | pieces.append(struct.pack('>cBi', b'D', decimals, raw)) |
| 120 | else: |
| 121 | # per spec, the "decimals" octet is unsigned (!) |
| 122 | pieces.append(struct.pack('>cBi', b'D', 0, int(value))) |
| 123 | return 6 |
| 124 | elif isinstance(value, datetime): |
| 125 | pieces.append( |
| 126 | struct.pack('>cQ', b'T', calendar.timegm(value.utctimetuple()))) |
| 127 | return 9 |
| 128 | elif isinstance(value, dict): |
| 129 | pieces.append(struct.pack('>c', b'F')) |
| 130 | return 1 + encode_table(pieces, value) |
| 131 | elif isinstance(value, list): |
| 132 | list_pieces: List[Any] = [] |
| 133 | for val in value: |
| 134 | encode_value(list_pieces, val) |
| 135 | piece = b''.join(list_pieces) |
| 136 | pieces.append(struct.pack('>cI', b'A', len(piece))) |
| 137 | pieces.append(piece) |
| 138 | return 5 + len(piece) |
no test coverage detected
searching dependent graphs…