Merge branch 'master' of github.com:mongodb/mongo-python-driver

Author: blink1073

doc/changelog.rst

@@ -9,6 +9,8 @@ PyMongo 4.12 brings a number of changes including:
 - Support for configuring DEK cache lifetime via the ``key_expiration_ms`` argument to
   :class:`~pymongo.encryption_options.AutoEncryptionOpts`.
 - Support for $lookup in CSFLE and QE supported on MongoDB 8.1+.
+- Added :meth:`gridfs.asynchronous.grid_file.AsyncGridFSBucket.delete_by_name` and :meth:`gridfs.grid_file.GridFSBucket.delete_by_name`
+  for more performant deletion of a file with multiple revisions.
 - AsyncMongoClient no longer performs DNS resolution for "mongodb+srv://" connection strings on creation.
   To avoid blocking the asyncio loop, the resolution is now deferred until the client is first connected.
 - Added index hinting support to the

gridfs/asynchronous/grid_file.py

@@ -834,6 +834,35 @@ async def delete(self, file_id: Any, session: Optional[AsyncClientSession] = Non
         if not res.deleted_count:
             raise NoFile("no file could be deleted because none matched %s" % file_id)
 
+    @_csot.apply
+    async def delete_by_name(
+        self, filename: str, session: Optional[AsyncClientSession] = None
+    ) -> None:
+        """Given a filename, delete this stored file's files collection document(s)
+        and associated chunks from a GridFS bucket.
+
+        For example::
+
+          my_db = AsyncMongoClient().test
+          fs = AsyncGridFSBucket(my_db)
+          await fs.upload_from_stream("test_file", "data I want to store!")
+          await fs.delete_by_name("test_file")
+
+        Raises :exc:`~gridfs.errors.NoFile` if no file with the given filename exists.
+
+        :param filename: The name of the file to be deleted.
+        :param session: a :class:`~pymongo.client_session.AsyncClientSession`
+
+        .. versionadded:: 4.12
+        """
+        _disallow_transactions(session)
+        files = self._files.find({"filename": filename}, {"_id": 1}, session=session)
+        file_ids = [file["_id"] async for file in files]
+        res = await self._files.delete_many({"_id": {"$in": file_ids}}, session=session)
+        await self._chunks.delete_many({"files_id": {"$in": file_ids}}, session=session)
+        if not res.deleted_count:
+            raise NoFile(f"no file could be deleted because none matched filename {filename!r}")
+
     def find(self, *args: Any, **kwargs: Any) -> AsyncGridOutCursor:
         """Find and return the files collection documents that match ``filter``
 

gridfs/synchronous/grid_file.py

@@ -830,6 +830,33 @@ def delete(self, file_id: Any, session: Optional[ClientSession] = None) -> None:
         if not res.deleted_count:
             raise NoFile("no file could be deleted because none matched %s" % file_id)
 
+    @_csot.apply
+    def delete_by_name(self, filename: str, session: Optional[ClientSession] = None) -> None:
+        """Given a filename, delete this stored file's files collection document(s)
+        and associated chunks from a GridFS bucket.
+
+        For example::
+
+          my_db = MongoClient().test
+          fs = GridFSBucket(my_db)
+          fs.upload_from_stream("test_file", "data I want to store!")
+          fs.delete_by_name("test_file")
+
+        Raises :exc:`~gridfs.errors.NoFile` if no file with the given filename exists.
+
+        :param filename: The name of the file to be deleted.
+        :param session: a :class:`~pymongo.client_session.ClientSession`
+
+        .. versionadded:: 4.12
+        """
+        _disallow_transactions(session)
+        files = self._files.find({"filename": filename}, {"_id": 1}, session=session)
+        file_ids = [file["_id"] for file in files]
+        res = self._files.delete_many({"_id": {"$in": file_ids}}, session=session)
+        self._chunks.delete_many({"files_id": {"$in": file_ids}}, session=session)
+        if not res.deleted_count:
+            raise NoFile(f"no file could be deleted because none matched filename {filename!r}")
+
     def find(self, *args: Any, **kwargs: Any) -> GridOutCursor:
         """Find and return the files collection documents that match ``filter``
 

pymongo/asynchronous/bulk.py

@@ -87,7 +87,7 @@ def __init__(
         self,
         collection: AsyncCollection[_DocumentType],
         ordered: bool,
-        bypass_document_validation: bool,
+        bypass_document_validation: Optional[bool],
         comment: Optional[str] = None,
         let: Optional[Any] = None,
     ) -> None:
@@ -516,8 +516,8 @@ async def _execute_command(
                 if self.comment:
                     cmd["comment"] = self.comment
                 _csot.apply_write_concern(cmd, write_concern)
-                if self.bypass_doc_val:
-                    cmd["bypassDocumentValidation"] = True
+                if self.bypass_doc_val is not None:
+                    cmd["bypassDocumentValidation"] = self.bypass_doc_val
                 if self.let is not None and run.op_type in (_DELETE, _UPDATE):
                     cmd["let"] = self.let
                 if session:

pymongo/asynchronous/collection.py

@@ -701,7 +701,7 @@ async def bulk_write(
         self,
         requests: Sequence[_WriteOp[_DocumentType]],
         ordered: bool = True,
-        bypass_document_validation: bool = False,
+        bypass_document_validation: Optional[bool] = None,
         session: Optional[AsyncClientSession] = None,
         comment: Optional[Any] = None,
         let: Optional[Mapping] = None,
@@ -800,7 +800,7 @@ async def _insert_one(
         ordered: bool,
         write_concern: WriteConcern,
         op_id: Optional[int],
-        bypass_doc_val: bool,
+        bypass_doc_val: Optional[bool],
         session: Optional[AsyncClientSession],
         comment: Optional[Any] = None,
     ) -> Any:
@@ -814,8 +814,8 @@ async def _insert_one(
         async def _insert_command(
             session: Optional[AsyncClientSession], conn: AsyncConnection, retryable_write: bool
         ) -> None:
-            if bypass_doc_val:
-                command["bypassDocumentValidation"] = True
+            if bypass_doc_val is not None:
+                command["bypassDocumentValidation"] = bypass_doc_val
 
             result = await conn.command(
                 self._database.name,
@@ -840,7 +840,7 @@ async def _insert_command(
     async def insert_one(
         self,
         document: Union[_DocumentType, RawBSONDocument],
-        bypass_document_validation: bool = False,
+        bypass_document_validation: Optional[bool] = None,
         session: Optional[AsyncClientSession] = None,
         comment: Optional[Any] = None,
     ) -> InsertOneResult:
@@ -906,7 +906,7 @@ async def insert_many(
         self,
         documents: Iterable[Union[_DocumentType, RawBSONDocument]],
         ordered: bool = True,
-        bypass_document_validation: bool = False,
+        bypass_document_validation: Optional[bool] = None,
         session: Optional[AsyncClientSession] = None,
         comment: Optional[Any] = None,
     ) -> InsertManyResult:
@@ -986,7 +986,7 @@ async def _update(
         write_concern: Optional[WriteConcern] = None,
         op_id: Optional[int] = None,
         ordered: bool = True,
-        bypass_doc_val: Optional[bool] = False,
+        bypass_doc_val: Optional[bool] = None,
         collation: Optional[_CollationIn] = None,
         array_filters: Optional[Sequence[Mapping[str, Any]]] = None,
         hint: Optional[_IndexKeyHint] = None,
@@ -1041,8 +1041,8 @@ async def _update(
         if comment is not None:
             command["comment"] = comment
         # Update command.
-        if bypass_doc_val:
-            command["bypassDocumentValidation"] = True
+        if bypass_doc_val is not None:
+            command["bypassDocumentValidation"] = bypass_doc_val
 
         # The command result has to be published for APM unmodified
         # so we make a shallow copy here before adding updatedExisting.
@@ -1082,7 +1082,7 @@ async def _update_retryable(
         write_concern: Optional[WriteConcern] = None,
         op_id: Optional[int] = None,
         ordered: bool = True,
-        bypass_doc_val: Optional[bool] = False,
+        bypass_doc_val: Optional[bool] = None,
         collation: Optional[_CollationIn] = None,
         array_filters: Optional[Sequence[Mapping[str, Any]]] = None,
         hint: Optional[_IndexKeyHint] = None,
@@ -1128,7 +1128,7 @@ async def replace_one(
         filter: Mapping[str, Any],
         replacement: Mapping[str, Any],
         upsert: bool = False,
-        bypass_document_validation: bool = False,
+        bypass_document_validation: Optional[bool] = None,
         collation: Optional[_CollationIn] = None,
         hint: Optional[_IndexKeyHint] = None,
         session: Optional[AsyncClientSession] = None,
@@ -1237,7 +1237,7 @@ async def update_one(
         filter: Mapping[str, Any],
         update: Union[Mapping[str, Any], _Pipeline],
         upsert: bool = False,
-        bypass_document_validation: bool = False,
+        bypass_document_validation: Optional[bool] = None,
         collation: Optional[_CollationIn] = None,
         array_filters: Optional[Sequence[Mapping[str, Any]]] = None,
         hint: Optional[_IndexKeyHint] = None,
@@ -2948,6 +2948,7 @@ async def aggregate(
             returning aggregate results using a cursor.
           - `collation` (optional): An instance of
             :class:`~pymongo.collation.Collation`.
+          - `bypassDocumentValidation` (bool): If ``True``, allows the write to opt-out of document level validation.
 
 
         :return: A :class:`~pymongo.asynchronous.command_cursor.AsyncCommandCursor` over the result

pymongo/asynchronous/encryption.py

@@ -64,11 +64,6 @@
 from pymongo.asynchronous.cursor import AsyncCursor
 from pymongo.asynchronous.database import AsyncDatabase
 from pymongo.asynchronous.mongo_client import AsyncMongoClient
-from pymongo.asynchronous.pool import (
-    _configured_socket,
-    _get_timeout_details,
-    _raise_connection_failure,
-)
 from pymongo.common import CONNECT_TIMEOUT
 from pymongo.daemon import _spawn_daemon
 from pymongo.encryption_options import AutoEncryptionOpts, RangeOpts
@@ -80,12 +75,17 @@
     NetworkTimeout,
     ServerSelectionTimeoutError,
 )
-from pymongo.network_layer import BLOCKING_IO_ERRORS, async_sendall
+from pymongo.network_layer import async_socket_sendall
 from pymongo.operations import UpdateOne
 from pymongo.pool_options import PoolOptions
+from pymongo.pool_shared import (
+    _async_configured_socket,
+    _get_timeout_details,
+    _raise_connection_failure,
+)
 from pymongo.read_concern import ReadConcern
 from pymongo.results import BulkWriteResult, DeleteResult
-from pymongo.ssl_support import get_ssl_context
+from pymongo.ssl_support import BLOCKING_IO_ERRORS, get_ssl_context
 from pymongo.typings import _DocumentType, _DocumentTypeArg
 from pymongo.uri_parser_shared import parse_host
 from pymongo.write_concern import WriteConcern
@@ -113,7 +113,7 @@
 
 async def _connect_kms(address: _Address, opts: PoolOptions) -> Union[socket.socket, _sslConn]:
     try:
-        return await _configured_socket(address, opts)
+        return await _async_configured_socket(address, opts)
     except Exception as exc:
         _raise_connection_failure(address, exc, timeout_details=_get_timeout_details(opts))
 
@@ -196,7 +196,7 @@ async def kms_request(self, kms_context: MongoCryptKmsContext) -> None:
         try:
             conn = await _connect_kms(address, opts)
             try:
-                await async_sendall(conn, message)
+                await async_socket_sendall(conn, message)
                 while kms_context.bytes_needed > 0:
                     # CSOT: update timeout.
                     conn.settimeout(max(_csot.clamp_remaining(_KMS_CONNECT_TIMEOUT), 0))

pymongo/asynchronous/mongo_client.py

@@ -2079,7 +2079,7 @@ async def _cleanup_cursor_lock(
                 # exhausted the result set we *must* close the socket
                 # to stop the server from sending more data.
                 assert conn_mgr.conn is not None
-                conn_mgr.conn.close_conn(ConnectionClosedReason.ERROR)
+                await conn_mgr.conn.close_conn(ConnectionClosedReason.ERROR)
             else:
                 await self._close_cursor_now(cursor_id, address, session=session, conn_mgr=conn_mgr)
         if conn_mgr:

pymongo/asynchronous/monitor.py

@@ -36,7 +36,11 @@
 from pymongo.server_description import ServerDescription
 
 if TYPE_CHECKING:
-    from pymongo.asynchronous.pool import AsyncConnection, Pool, _CancellationContext
+    from pymongo.asynchronous.pool import (  # type: ignore[attr-defined]
+        AsyncConnection,
+        Pool,
+        _CancellationContext,
+    )
     from pymongo.asynchronous.settings import TopologySettings
     from pymongo.asynchronous.topology import Topology
 

pymongo/asynchronous/network.py

@@ -17,7 +17,6 @@
 
 import datetime
 import logging
-import time
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -31,20 +30,16 @@
 
 from bson import _decode_all_selective
 from pymongo import _csot, helpers_shared, message
-from pymongo.common import MAX_MESSAGE_SIZE
-from pymongo.compression_support import _NO_COMPRESSION, decompress
+from pymongo.compression_support import _NO_COMPRESSION
 from pymongo.errors import (
     NotPrimaryError,
     OperationFailure,
-    ProtocolError,
 )
 from pymongo.logger import _COMMAND_LOGGER, _CommandStatusMessage, _debug_log
-from pymongo.message import _UNPACK_REPLY, _OpMsg, _OpReply
+from pymongo.message import _OpMsg
 from pymongo.monitoring import _is_speculative_authenticate
 from pymongo.network_layer import (
-    _UNPACK_COMPRESSION_HEADER,
-    _UNPACK_HEADER,
-    async_receive_data,
+    async_receive_message,
     async_sendall,
 )
 
@@ -194,13 +189,13 @@ async def command(
         )
 
     try:
-        await async_sendall(conn.conn, msg)
+        await async_sendall(conn.conn.get_conn, msg)
         if use_op_msg and unacknowledged:
             # Unacknowledged, fake a successful command response.
             reply = None
             response_doc: _DocumentOut = {"ok": 1}
         else:
-            reply = await receive_message(conn, request_id)
+            reply = await async_receive_message(conn, request_id)
             conn.more_to_come = reply.more_to_come
             unpacked_docs = reply.unpack_response(
                 codec_options=codec_options, user_fields=user_fields
@@ -301,47 +296,3 @@ async def command(
         )
 
     return response_doc  # type: ignore[return-value]
-
-
-async def receive_message(
-    conn: AsyncConnection, request_id: Optional[int], max_message_size: int = MAX_MESSAGE_SIZE
-) -> Union[_OpReply, _OpMsg]:
-    """Receive a raw BSON message or raise socket.error."""
-    if _csot.get_timeout():
-        deadline = _csot.get_deadline()
-    else:
-        timeout = conn.conn.gettimeout()
-        if timeout:
-            deadline = time.monotonic() + timeout
-        else:
-            deadline = None
-    # Ignore the response's request id.
-    length, _, response_to, op_code = _UNPACK_HEADER(await async_receive_data(conn, 16, deadline))
-    # No request_id for exhaust cursor "getMore".
-    if request_id is not None:
-        if request_id != response_to:
-            raise ProtocolError(f"Got response id {response_to!r} but expected {request_id!r}")
-    if length <= 16:
-        raise ProtocolError(
-            f"Message length ({length!r}) not longer than standard message header size (16)"
-        )
-    if length > max_message_size:
-        raise ProtocolError(
-            f"Message length ({length!r}) is larger than server max "
-            f"message size ({max_message_size!r})"
-        )
-    if op_code == 2012:
-        op_code, _, compressor_id = _UNPACK_COMPRESSION_HEADER(
-            await async_receive_data(conn, 9, deadline)
-        )
-        data = decompress(await async_receive_data(conn, length - 25, deadline), compressor_id)
-    else:
-        data = await async_receive_data(conn, length - 16, deadline)
-
-    try:
-        unpack_reply = _UNPACK_REPLY[op_code]
-    except KeyError:
-        raise ProtocolError(
-            f"Got opcode {op_code!r} but expected {_UNPACK_REPLY.keys()!r}"
-        ) from None
-    return unpack_reply(data)