Flush the file to the database.
(self)
| 1301 | self._buffer = io.BytesIO() |
| 1302 | |
| 1303 | async def _flush(self) -> Any: |
| 1304 | """Flush the file to the database.""" |
| 1305 | try: |
| 1306 | await self._flush_buffer(force=True) |
| 1307 | # The GridFS spec says length SHOULD be an Int64. |
| 1308 | self._file["length"] = Int64(self._position) |
| 1309 | self._file["uploadDate"] = datetime.datetime.now(tz=datetime.timezone.utc) |
| 1310 | |
| 1311 | return await self._coll.files.insert_one(self._file, session=self._session) |
| 1312 | except DuplicateKeyError: |
| 1313 | self._raise_file_exists(self._id) |
| 1314 | |
| 1315 | def _raise_file_exists(self, file_id: Any) -> NoReturn: |
| 1316 | """Raise a FileExists exception for the given file_id.""" |
no test coverage detected