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