Method __init__

pymongo/synchronous/encryption.py:139–164

Internal class to perform I/O on behalf of pymongocrypt.

(
        self,
        client: Optional[MongoClient[_DocumentTypeArg]],
        key_vault_coll: Collection[_DocumentTypeArg],
        mongocryptd_client: Optional[MongoClient[_DocumentTypeArg]],
        opts: AutoEncryptionOpts,
    )

class _EncryptionIO(MongoCryptCallback):  # type: ignore[misc]
    def __init__(
        self,
        client: Optional[MongoClient[_DocumentTypeArg]],
        key_vault_coll: Collection[_DocumentTypeArg],
        mongocryptd_client: Optional[MongoClient[_DocumentTypeArg]],
        opts: AutoEncryptionOpts,
    ):
        """Internal class to perform I/O on behalf of pymongocrypt."""
        self.client_ref: Any
        # Use a weak ref to break reference cycle.
        if client is not None:
            self.client_ref = weakref.ref(client)
        else:
            self.client_ref = None
        self.key_vault_coll: Optional[Collection[RawBSONDocument]] = cast(
            Collection[RawBSONDocument],
            key_vault_coll.with_options(
                codec_options=_KEY_VAULT_OPTS,
                read_concern=ReadConcern(level="majority"),
                write_concern=WriteConcern(w="majority"),
            ),
        )
        self.mongocryptd_client = mongocryptd_client
        self.opts = opts
        self._spawned = False
        self._kms_ssl_contexts = opts._kms_ssl_contexts(_IS_SYNC)

    def kms_request(self, kms_context: MongoCryptKmsContext) -> None:
        """Complete a KMS request.