Create a batched OP_QUERY write command.
(
namespace: str,
operation: int,
command: MutableMapping[str, Any],
docs: list[Mapping[str, Any]],
opts: CodecOptions[Any],
ctx: _BulkWriteContext,
buf: _BytesIO,
)
| 1268 | |
| 1269 | |
| 1270 | def _batched_write_command_impl( |
| 1271 | namespace: str, |
| 1272 | operation: int, |
| 1273 | command: MutableMapping[str, Any], |
| 1274 | docs: list[Mapping[str, Any]], |
| 1275 | opts: CodecOptions[Any], |
| 1276 | ctx: _BulkWriteContext, |
| 1277 | buf: _BytesIO, |
| 1278 | ) -> tuple[list[Mapping[str, Any]], int]: |
| 1279 | """Create a batched OP_QUERY write command.""" |
| 1280 | max_bson_size = ctx.max_bson_size |
| 1281 | max_write_batch_size = ctx.max_write_batch_size |
| 1282 | # Max BSON object size + 16k - 2 bytes for ending NUL bytes. |
| 1283 | # Server guarantees there is enough room: SERVER-10643. |
| 1284 | max_cmd_size = max_bson_size + _COMMAND_OVERHEAD |
| 1285 | max_split_size = ctx.max_split_size |
| 1286 | |
| 1287 | # No options |
| 1288 | buf.write(_ZERO_32) |
| 1289 | # Namespace as C string |
| 1290 | buf.write(namespace.encode("utf8")) |
| 1291 | buf.write(_ZERO_8) |
| 1292 | # Skip: 0, Limit: -1 |
| 1293 | buf.write(_SKIPLIM) |
| 1294 | |
| 1295 | # Where to write command document length |
| 1296 | command_start = buf.tell() |
| 1297 | buf.write(bson.encode(command)) |
| 1298 | |
| 1299 | # Start of payload |
| 1300 | buf.seek(-1, 2) |
| 1301 | try: |
| 1302 | buf.write(_OP_MAP[operation]) |
| 1303 | except KeyError: |
| 1304 | raise InvalidOperation("Unknown command") from None |
| 1305 | |
| 1306 | # Where to write list document length |
| 1307 | list_start = buf.tell() - 4 |
| 1308 | to_send = [] |
| 1309 | idx = 0 |
| 1310 | for doc in docs: |
| 1311 | # Encode the current operation |
| 1312 | key = str(idx).encode("utf8") |
| 1313 | value = _dict_to_bson(doc, False, opts) |
| 1314 | # Is there enough room to add this document? max_cmd_size accounts for |
| 1315 | # the two trailing null bytes. |
| 1316 | doc_too_large = len(value) > max_cmd_size |
| 1317 | if doc_too_large: |
| 1318 | write_op = list(_FIELD_MAP.keys())[operation] |
| 1319 | _raise_document_too_large(write_op, len(value), max_bson_size) |
| 1320 | enough_data = idx >= 1 and (buf.tell() + len(key) + len(value)) >= max_split_size |
| 1321 | enough_documents = idx >= max_write_batch_size |
| 1322 | if enough_data or enough_documents: |
| 1323 | break |
| 1324 | buf.write(_BSONOBJ) |
| 1325 | buf.write(key) |
| 1326 | buf.write(_ZERO_8) |
| 1327 | buf.write(value) |
no test coverage detected