Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
session.error_wrapper = lambda e: NvimError(decode_if_bytes(e[1]))
channel_id, metadata = session.request(b'nvim_get_api_info')
self.metadata = metadata
version = metadata.get("version", {"api_level": 0})
self.version = Version(**version)
self.types = types
self.api = RemoteApi(self, 'nvim_')
self.vars = RemoteMap(self, 'nvim_get_var', 'nvim_set_var', 'nvim_del_var')
self.vvars = RemoteMap(self, 'nvim_get_vvar', None, None)
self.options = RemoteMap(self, 'nvim_get_option', 'nvim_set_option')
self.buffers = Buffers(self)
self.windows = RemoteSequence(self, 'nvim_list_wins')
self.tabpages = RemoteSequence(self, 'nvim_list_tabpages')
self.current = Current(self)
self.session = CompatibilitySession(self)
self.funcs = Funcs(self)
self.lua = LuaFuncs(self)
self.error = NvimError
self._decode = decode
self._err_cb = err_cb
# only on python3.4+ we expose asyncio
if IS_PYTHON3:
self.loop = self._session.loop._loop
def cleanup(self):
"""Cleanup the windows and buffers."""
for buf in self.buffers:
try:
self.vim.command(f"silent bdelete {buf.handle}")
except pynvim.api.common.NvimError as ex:
self.logger.warning("Skip cleaning up the buffer: %s", ex)
Normally a blocking request will be sent. If the `async_` flag is
present and True, a asynchronous notification is sent instead. This
will never block, and the return value or error is ignored.
"""
if (self._session._loop_thread is not None
and threading.current_thread() != self._session._loop_thread):
msg = ("Request from non-main thread.\n"
"Requests from different threads should be wrapped "
"with nvim.async_call(cb, ...) \n{}\n"
.format('\n'.join(format_stack(None, 5)[:-1])))
self.async_call(self._err_cb, msg)
raise NvimError("request from non-main thread")
decode = kwargs.pop('decode', self._decode)
args = walk(self._to_nvim, args)
res = self._session.request(name, *args, **kwargs)
return walk(self._from_nvim, res, decode=decode)