Function _connect_kms

pymongo/synchronous/encryption.py:118–122

(address: _Address, opts: PoolOptions)

def _connect_kms(address: _Address, opts: PoolOptions) -> Union[socket.socket, _sslConn]:
    try:
        return _configured_socket(address, opts)
    except Exception as exc:
        _raise_connection_failure(address, exc, timeout_details=_get_timeout_details(opts))


@contextlib.contextmanager