Representation of a MySQL response packet. Provides an interface for reading/parsing the packet results.
| 44 | |
| 45 | |
| 46 | class MysqlPacket: |
| 47 | """Representation of a MySQL response packet. |
| 48 | |
| 49 | Provides an interface for reading/parsing the packet results. |
| 50 | """ |
| 51 | |
| 52 | __slots__ = ("_data", "_position") |
| 53 | |
| 54 | def __init__(self, data, encoding): |
| 55 | self._position = 0 |
| 56 | self._data = data |
| 57 | |
| 58 | def get_all_data(self): |
| 59 | return self._data |
| 60 | |
| 61 | def read(self, size): |
| 62 | """Read the first 'size' bytes in packet and advance cursor past them.""" |
| 63 | result = self._data[self._position : (self._position + size)] |
| 64 | if len(result) != size: |
| 65 | error = ( |
| 66 | "Result length not requested length:\n" |
| 67 | f"Expected={size}. Actual={len(result)}. Position: {self._position}. Data Length: {len(self._data)}" |
| 68 | ) |
| 69 | if DEBUG: |
| 70 | print(error) |
| 71 | self.dump() |
| 72 | raise AssertionError(error) |
| 73 | self._position += size |
| 74 | return result |
| 75 | |
| 76 | def read_all(self): |
| 77 | """Read all remaining data in the packet. |
| 78 | |
| 79 | (Subsequent read() will return errors.) |
| 80 | """ |
| 81 | result = self._data[self._position :] |
| 82 | self._position = None # ensure no subsequent read() |
| 83 | return result |
| 84 | |
| 85 | def advance(self, length): |
| 86 | """Advance the cursor in data buffer 'length' bytes.""" |
| 87 | new_position = self._position + length |
| 88 | if new_position < 0 or new_position > len(self._data): |
| 89 | raise Exception( |
| 90 | f"Invalid advance amount ({length}) for cursor. Position={new_position}" |
| 91 | ) |
| 92 | self._position = new_position |
| 93 | |
| 94 | def rewind(self, position=0): |
| 95 | """Set the position of the data buffer cursor to 'position'.""" |
| 96 | if position < 0 or position > len(self._data): |
| 97 | raise Exception("Invalid position to rewind cursor to: %s." % position) |
| 98 | self._position = position |
| 99 | |
| 100 | def get_bytes(self, position, length=1): |
| 101 | """Get 'length' bytes starting at 'position'. |
| 102 | |
| 103 | Position is start of payload (first four packet header bytes are not |