Create a _Encrypter for a client. :param client: The encrypted MongoClient. :param opts: The encrypted client's :class:`AutoEncryptionOpts`.
(self, client: MongoClient[_DocumentTypeArg], opts: AutoEncryptionOpts)
| 385 | """ |
| 386 | |
| 387 | def __init__(self, client: MongoClient[_DocumentTypeArg], opts: AutoEncryptionOpts): |
| 388 | """Create a _Encrypter for a client. |
| 389 | |
| 390 | :param client: The encrypted MongoClient. |
| 391 | :param opts: The encrypted client's :class:`AutoEncryptionOpts`. |
| 392 | """ |
| 393 | if opts._schema_map is None: |
| 394 | schema_map = None |
| 395 | else: |
| 396 | schema_map = _dict_to_bson(opts._schema_map, False, _DATA_KEY_OPTS) |
| 397 | |
| 398 | if opts._encrypted_fields_map is None: |
| 399 | encrypted_fields_map = None |
| 400 | else: |
| 401 | encrypted_fields_map = _dict_to_bson(opts._encrypted_fields_map, False, _DATA_KEY_OPTS) |
| 402 | self._bypass_auto_encryption = opts._bypass_auto_encryption |
| 403 | self._internal_client = None |
| 404 | # parsing kms_ssl_contexts here so that parsing errors will be raised before internal clients are created |
| 405 | opts._kms_ssl_contexts(_IS_SYNC) |
| 406 | |
| 407 | def _get_internal_client( |
| 408 | encrypter: _Encrypter, mongo_client: MongoClient[_DocumentTypeArg] |
| 409 | ) -> MongoClient[_DocumentTypeArg]: |
| 410 | if mongo_client.options.pool_options.max_pool_size is None: |
| 411 | # Unlimited pool size, use the same client. |
| 412 | return mongo_client |
| 413 | # Else - limited pool size, use an internal client. |
| 414 | if encrypter._internal_client is not None: |
| 415 | return encrypter._internal_client |
| 416 | internal_client = mongo_client._duplicate(minPoolSize=0, auto_encryption_opts=None) |
| 417 | encrypter._internal_client = internal_client |
| 418 | return internal_client |
| 419 | |
| 420 | if opts._key_vault_client is not None: |
| 421 | key_vault_client = opts._key_vault_client |
| 422 | else: |
| 423 | key_vault_client = _get_internal_client(self, client) |
| 424 | |
| 425 | if opts._bypass_auto_encryption: |
| 426 | metadata_client = None |
| 427 | else: |
| 428 | metadata_client = _get_internal_client(self, client) |
| 429 | |
| 430 | db, coll = opts._key_vault_namespace.split(".", 1) |
| 431 | key_vault_coll = key_vault_client[db][coll] |
| 432 | |
| 433 | mongocryptd_client: MongoClient[Mapping[str, Any]] = MongoClient( |
| 434 | opts._mongocryptd_uri, connect=False, serverSelectionTimeoutMS=_MONGOCRYPTD_TIMEOUT_MS |
| 435 | ) |
| 436 | |
| 437 | io_callbacks = _EncryptionIO( # type:ignore[misc] |
| 438 | metadata_client, |
| 439 | key_vault_coll, # type:ignore[arg-type] |
| 440 | mongocryptd_client, |
| 441 | opts, |
| 442 | ) |
| 443 | self._auto_encrypter = AutoEncrypter( |
| 444 | io_callbacks, |
nothing calls this directly
no test coverage detected