Method mark_command

pymongo/synchronous/encryption.py:277–302

Mark a command for encryption. :param database: The database on which to run this command. :param cmd: The BSON command to run. :return: The marked command response from mongocryptd.

(self, database: str, cmd: bytes)

        _spawn_daemon(args)

    def mark_command(self, database: str, cmd: bytes) -> bytes | memoryview:
        """Mark a command for encryption.

        :param database: The database on which to run this command.
        :param cmd: The BSON command to run.

        :return: The marked command response from mongocryptd.
        """
        if not self._spawned and not self.opts._mongocryptd_bypass_spawn:
            self.spawn()
        # Database.command only supports mutable mappings so we need to decode
        # the raw BSON command first.
        inflated_cmd = _inflate_bson(cmd, DEFAULT_RAW_BSON_OPTIONS)
        assert self.mongocryptd_client is not None
        try:
            res = self.mongocryptd_client[database].command(
                inflated_cmd, codec_options=DEFAULT_RAW_BSON_OPTIONS
            )
        except ServerSelectionTimeoutError:
            if self.opts._mongocryptd_bypass_spawn:
                raise
            self.spawn()
            res = self.mongocryptd_client[database].command(
                inflated_cmd, codec_options=DEFAULT_RAW_BSON_OPTIONS
            )
        return res.raw

    def fetch_keys(self, filter: bytes) -> Generator[bytes | memoryview, None]:
        """Yields one or more keys from the key vault.