Flush `data` to a chunk.
(self, data: Any, force: bool = False)
| 1247 | self._coll.files.update_one({"_id": self._file["_id"]}, {"$set": {name: value}}) |
| 1248 | |
| 1249 | def _flush_data(self, data: Any, force: bool = False) -> None: |
| 1250 | """Flush `data` to a chunk.""" |
| 1251 | self._ensure_indexes() |
| 1252 | assert len(data) <= self.chunk_size |
| 1253 | if data: |
| 1254 | self._buffered_docs.append( |
| 1255 | {"files_id": self._file["_id"], "n": self._chunk_number, "data": data} |
| 1256 | ) |
| 1257 | self._buffered_docs_size += len(data) + _CHUNK_OVERHEAD |
| 1258 | if not self._buffered_docs: |
| 1259 | return |
| 1260 | # Limit to 100,000 chunks or 32MB (+1 chunk) of data. |
| 1261 | if ( |
| 1262 | force |
| 1263 | or self._buffered_docs_size >= _UPLOAD_BUFFER_SIZE |
| 1264 | or len(self._buffered_docs) >= _UPLOAD_BUFFER_CHUNKS |
| 1265 | ): |
| 1266 | try: |
| 1267 | self._chunks.insert_many(self._buffered_docs, session=self._session) |
| 1268 | except BulkWriteError as exc: |
| 1269 | # For backwards compatibility, raise an insert_one style exception. |
| 1270 | write_errors = exc.details["writeErrors"] |
| 1271 | for err in write_errors: |
| 1272 | if err.get("code") in (11000, 11001, 12582): # Duplicate key errors |
| 1273 | self._raise_file_exists(self._file["_id"]) |
| 1274 | result = {"writeErrors": write_errors} |
| 1275 | wces = exc.details["writeConcernErrors"] |
| 1276 | if wces: |
| 1277 | result["writeConcernError"] = wces[-1] |
| 1278 | _check_write_command_response(result) |
| 1279 | raise |
| 1280 | self._buffered_docs = [] |
| 1281 | self._buffered_docs_size = 0 |
| 1282 | self._chunk_number += 1 |
| 1283 | self._position += len(data) |
| 1284 | |
| 1285 | def _flush_buffer(self, force: bool = False) -> None: |
| 1286 | """Flush the buffer contents out to a chunk.""" |
no test coverage detected