(
self, name: str, session: Optional[AsyncClientSession] = None, comment: Optional[Any] = None
)
| 1257 | return await self._list_collection_names(session, filter, comment, **kwargs) |
| 1258 | |
| 1259 | async def _drop_helper( |
| 1260 | self, name: str, session: Optional[AsyncClientSession] = None, comment: Optional[Any] = None |
| 1261 | ) -> dict[str, Any]: |
| 1262 | command = {"drop": name} |
| 1263 | if comment is not None: |
| 1264 | command["comment"] = comment |
| 1265 | |
| 1266 | async def inner( |
| 1267 | session: Optional[AsyncClientSession], conn: AsyncConnection, _retryable_write: bool |
| 1268 | ) -> dict[str, Any]: |
| 1269 | return await self._command( |
| 1270 | conn, |
| 1271 | command, |
| 1272 | allowable_errors=["ns not found", 26], |
| 1273 | write_concern=self._write_concern_for(session), |
| 1274 | parse_write_concern_error=True, |
| 1275 | session=session, |
| 1276 | ) |
| 1277 | |
| 1278 | return await self.client._retryable_write(False, inner, session, _Op.DROP) |
| 1279 | |
| 1280 | @_csot.apply |
| 1281 | async def drop_collection( |
no test coverage detected