Check the response header from the database, without decoding BSON. Check the response for errors and unpack. Can raise CursorNotFound, NotPrimaryError, ExecutionTimeout, or OperationFailure. :param cursor_id: cursor_id we sent to get this response - us
(
self, cursor_id: Optional[int] = None, user_fields: Optional[Mapping[str, Any]] = None
)
| 1359 | self.documents = documents |
| 1360 | |
| 1361 | def raw_response( |
| 1362 | self, cursor_id: Optional[int] = None, user_fields: Optional[Mapping[str, Any]] = None |
| 1363 | ) -> list[bytes | memoryview]: |
| 1364 | """Check the response header from the database, without decoding BSON. |
| 1365 | |
| 1366 | Check the response for errors and unpack. |
| 1367 | |
| 1368 | Can raise CursorNotFound, NotPrimaryError, ExecutionTimeout, or |
| 1369 | OperationFailure. |
| 1370 | |
| 1371 | :param cursor_id: cursor_id we sent to get this response - |
| 1372 | used for raising an informative exception when we get cursor id not |
| 1373 | valid at server response. |
| 1374 | """ |
| 1375 | if self.flags & 1: |
| 1376 | # Shouldn't get this response if we aren't doing a getMore |
| 1377 | if cursor_id is None: |
| 1378 | raise ProtocolError("No cursor id for getMore operation") |
| 1379 | |
| 1380 | # Fake a getMore command response. OP_GET_MORE provides no |
| 1381 | # document. |
| 1382 | msg = "Cursor not found, cursor id: %d" % (cursor_id,) |
| 1383 | errobj = {"ok": 0, "errmsg": msg, "code": 43} |
| 1384 | raise CursorNotFound(msg, 43, errobj) |
| 1385 | elif self.flags & 2: |
| 1386 | error_object: dict[str, Any] = bson.BSON(self.documents).decode() |
| 1387 | # Fake the ok field if it doesn't exist. |
| 1388 | error_object.setdefault("ok", 0) |
| 1389 | if error_object["$err"].startswith(HelloCompat.LEGACY_ERROR): |
| 1390 | raise NotPrimaryError(error_object["$err"], error_object) |
| 1391 | elif error_object.get("code") == 50: |
| 1392 | default_msg = "operation exceeded time limit" |
| 1393 | raise ExecutionTimeout( |
| 1394 | error_object.get("$err", default_msg), error_object.get("code"), error_object |
| 1395 | ) |
| 1396 | raise OperationFailure( |
| 1397 | "database error: %s" % error_object.get("$err"), |
| 1398 | error_object.get("code"), |
| 1399 | error_object, |
| 1400 | ) |
| 1401 | if self.documents: |
| 1402 | return [self.documents] |
| 1403 | return [] |
| 1404 | |
| 1405 | def unpack_response( |
| 1406 | self, |
no test coverage detected