Flush the file to the database.
(self)
| 1289 | self._buffer = io.BytesIO() |
| 1290 | |
| 1291 | def _flush(self) -> Any: |
| 1292 | """Flush the file to the database.""" |
| 1293 | try: |
| 1294 | self._flush_buffer(force=True) |
| 1295 | # The GridFS spec says length SHOULD be an Int64. |
| 1296 | self._file["length"] = Int64(self._position) |
| 1297 | self._file["uploadDate"] = datetime.datetime.now(tz=datetime.timezone.utc) |
| 1298 | |
| 1299 | return self._coll.files.insert_one(self._file, session=self._session) |
| 1300 | except DuplicateKeyError: |
| 1301 | self._raise_file_exists(self._id) |
| 1302 | |
| 1303 | def _raise_file_exists(self, file_id: Any) -> NoReturn: |
| 1304 | """Raise a FileExists exception for the given file_id.""" |
no test coverage detected