Encode an error response message. Args: request_id: Request identifier this responds to error: DirtyError instance, dict, or Exception Returns: bytes: Complete message (header + payload)
(request_id: int, error)
| 223 | |
| 224 | @staticmethod |
| 225 | def encode_error(request_id: int, error) -> bytes: |
| 226 | """ |
| 227 | Encode an error response message. |
| 228 | |
| 229 | Args: |
| 230 | request_id: Request identifier this responds to |
| 231 | error: DirtyError instance, dict, or Exception |
| 232 | |
| 233 | Returns: |
| 234 | bytes: Complete message (header + payload) |
| 235 | """ |
| 236 | from .errors import DirtyError |
| 237 | |
| 238 | if isinstance(error, DirtyError): |
| 239 | error_dict = error.to_dict() |
| 240 | elif isinstance(error, dict): |
| 241 | error_dict = error |
| 242 | else: |
| 243 | error_dict = { |
| 244 | "error_type": type(error).__name__, |
| 245 | "message": str(error), |
| 246 | "details": {}, |
| 247 | } |
| 248 | |
| 249 | payload_dict = {"error": error_dict} |
| 250 | payload = TLVEncoder.encode(payload_dict) |
| 251 | header = BinaryProtocol.encode_header(MSG_TYPE_ERROR, request_id, |
| 252 | len(payload)) |
| 253 | return header + payload |
| 254 | |
| 255 | @staticmethod |
| 256 | def encode_chunk(request_id: int, data) -> bytes: |