Internal class to perform I/O on behalf of pymongocrypt.
(
self,
client: Optional[AsyncMongoClient[_DocumentTypeArg]],
key_vault_coll: AsyncCollection[_DocumentTypeArg],
mongocryptd_client: Optional[AsyncMongoClient[_DocumentTypeArg]],
opts: AutoEncryptionOpts,
)
| 138 | |
| 139 | class _EncryptionIO(AsyncMongoCryptCallback): # type: ignore[misc] |
| 140 | def __init__( |
| 141 | self, |
| 142 | client: Optional[AsyncMongoClient[_DocumentTypeArg]], |
| 143 | key_vault_coll: AsyncCollection[_DocumentTypeArg], |
| 144 | mongocryptd_client: Optional[AsyncMongoClient[_DocumentTypeArg]], |
| 145 | opts: AutoEncryptionOpts, |
| 146 | ): |
| 147 | """Internal class to perform I/O on behalf of pymongocrypt.""" |
| 148 | self.client_ref: Any |
| 149 | # Use a weak ref to break reference cycle. |
| 150 | if client is not None: |
| 151 | self.client_ref = weakref.ref(client) |
| 152 | else: |
| 153 | self.client_ref = None |
| 154 | self.key_vault_coll: Optional[AsyncCollection[RawBSONDocument]] = cast( |
| 155 | AsyncCollection[RawBSONDocument], |
| 156 | key_vault_coll.with_options( |
| 157 | codec_options=_KEY_VAULT_OPTS, |
| 158 | read_concern=ReadConcern(level="majority"), |
| 159 | write_concern=WriteConcern(w="majority"), |
| 160 | ), |
| 161 | ) |
| 162 | self.mongocryptd_client = mongocryptd_client |
| 163 | self.opts = opts |
| 164 | self._spawned = False |
| 165 | self._kms_ssl_contexts = opts._kms_ssl_contexts(_IS_SYNC) |
| 166 | |
| 167 | async def kms_request(self, kms_context: MongoCryptKmsContext) -> None: |
| 168 | """Complete a KMS request. |
nothing calls this directly
no test coverage detected