Sends a create command with the given options.
(
self,
name: str,
options: MutableMapping[str, Any],
collation: Optional[_CollationIn],
session: Optional[ClientSession],
encrypted_fields: Optional[Mapping[str, Any]] = None,
qev2_required: bool = False,
)
| 630 | ) |
| 631 | |
| 632 | def _create_helper( |
| 633 | self, |
| 634 | name: str, |
| 635 | options: MutableMapping[str, Any], |
| 636 | collation: Optional[_CollationIn], |
| 637 | session: Optional[ClientSession], |
| 638 | encrypted_fields: Optional[Mapping[str, Any]] = None, |
| 639 | qev2_required: bool = False, |
| 640 | ) -> None: |
| 641 | """Sends a create command with the given options.""" |
| 642 | cmd: dict[str, Any] = {"create": name} |
| 643 | if encrypted_fields: |
| 644 | cmd["encryptedFields"] = encrypted_fields |
| 645 | |
| 646 | if options: |
| 647 | if "size" in options: |
| 648 | options["size"] = float(options["size"]) |
| 649 | cmd.update(options) |
| 650 | |
| 651 | def inner( |
| 652 | session: Optional[ClientSession], conn: Connection, _retryable_write: bool |
| 653 | ) -> None: |
| 654 | if qev2_required and conn.max_wire_version < 21: |
| 655 | raise ConfigurationError( |
| 656 | "Driver support of Queryable Encryption is incompatible with server. " |
| 657 | "Upgrade server to use Queryable Encryption. " |
| 658 | f"Got maxWireVersion {conn.max_wire_version} but need maxWireVersion >= 21 (MongoDB >=7.0)" |
| 659 | ) |
| 660 | |
| 661 | self._command( |
| 662 | conn, |
| 663 | cmd, |
| 664 | read_preference=ReadPreference.PRIMARY, |
| 665 | write_concern=self._write_concern_for(session), |
| 666 | collation=collation, |
| 667 | session=session, |
| 668 | ) |
| 669 | |
| 670 | self.database.client._retryable_write(False, inner, session, _Op.CREATE) |
| 671 | |
| 672 | def _create( |
| 673 | self, |
no test coverage detected