1 """Main Nvim interface."""
5 from functools import partial
6 from traceback import format_stack
8 from msgpack import ExtType
10 from pynvim.api.buffer import Buffer
11 from pynvim.api.common import (NvimError, Remote, RemoteApi, RemoteMap, RemoteSequence,
12 decode_if_bytes, walk)
13 from pynvim.api.tabpage import Tabpage
14 from pynvim.api.window import Window
15 from pynvim.compat import IS_PYTHON3
16 from pynvim.util import Version, format_exc_skip
25 local function update_highlights(buf, src_id, hls, clear_first, clear_end)
26 if clear_first ~= nil then
27 a.nvim_buf_clear_highlight(buf, src_id, clear_first, clear_end)
29 for _,hl in pairs(hls) do
30 local group, line, col_start, col_end = unpack(hl)
31 if col_start == nil then
34 if col_end == nil then
37 a.nvim_buf_add_highlight(buf, src_id, group, line, col_start, col_end)
42 local mod = {update_highlights=update_highlights}
43 _G["_pynvim_"..chid] = mod
49 """Class that represents a remote Nvim instance.
51 This class is main entry point to Nvim remote API, it is a wrapper
52 around Session instances.
54 The constructor of this class must not be called directly. Instead, the
55 `from_session` class method should be used to create the first instance
56 from a raw `Session` instance.
58 Subsequent instances for the same session can be created by calling the
59 `with_decode` instance method to change the decoding behavior or
60 `SubClass.from_nvim(nvim)` where `SubClass` is a subclass of `Nvim`, which
61 is useful for having multiple `Nvim` objects that behave differently
62 without one affecting the other.
64 When this library is used on python3.4+, asyncio event loop is guaranteed
65 to be used. It is available as the "loop" attribute of this class. Note
66 that asyncio callbacks cannot make blocking requests, which includes
67 accessing state-dependent attributes. They should instead schedule another
68 callback using nvim.async_call, which will not have this restriction.
72 def from_session(cls, session):
73 """Create a new Nvim instance for a Session instance.
75 This method must be called to create the first Nvim instance, since it
76 queries Nvim metadata for type information and sets a SessionHook for
77 creating specialized objects from Nvim remote handles.
79 session.error_wrapper = lambda e: NvimError(decode_if_bytes(e[1]))
80 channel_id, metadata = session.request(b'nvim_get_api_info')
83 # decode all metadata strings for python3
84 metadata = walk(decode_if_bytes, metadata)
87 metadata['types']['Buffer']['id']: Buffer,
88 metadata['types']['Window']['id']: Window,
89 metadata['types']['Tabpage']['id']: Tabpage,
92 return cls(session, channel_id, metadata, types)
95 def from_nvim(cls, nvim):
96 """Create a new Nvim instance from an existing instance."""
97 return cls(nvim._session, nvim.channel_id, nvim.metadata,
98 nvim.types, nvim._decode, nvim._err_cb)
100 def __init__(self, session, channel_id, metadata, types,
101 decode=False, err_cb=None):
102 """Initialize a new Nvim instance. This method is module-private."""
103 self._session = session
104 self.channel_id = channel_id
105 self.metadata = metadata
106 version = metadata.get("version", {"api_level": 0})
107 self.version = Version(**version)
109 self.api = RemoteApi(self, 'nvim_')
110 self.vars = RemoteMap(self, 'nvim_get_var', 'nvim_set_var', 'nvim_del_var')
111 self.vvars = RemoteMap(self, 'nvim_get_vvar', None, None)
112 self.options = RemoteMap(self, 'nvim_get_option', 'nvim_set_option')
113 self.buffers = Buffers(self)
114 self.windows = RemoteSequence(self, 'nvim_list_wins')
115 self.tabpages = RemoteSequence(self, 'nvim_list_tabpages')
116 self.current = Current(self)
117 self.session = CompatibilitySession(self)
118 self.funcs = Funcs(self)
119 self.lua = LuaFuncs(self)
120 self.error = NvimError
121 self._decode = decode
122 self._err_cb = err_cb
124 # only on python3.4+ we expose asyncio
126 self.loop = self._session.loop._loop
128 def _from_nvim(self, obj, decode=None):
130 decode = self._decode
131 if type(obj) is ExtType:
132 cls = self.types[obj.code]
133 return cls(self, (obj.code, obj.data))
135 obj = decode_if_bytes(obj, decode)
138 def _to_nvim(self, obj):
139 if isinstance(obj, Remote):
140 return ExtType(*obj.code_data)
143 def _get_lua_private(self):
144 if not getattr(self._session, "_has_lua", False):
145 self.exec_lua(lua_module, self.channel_id)
146 self._session._has_lua = True
147 return getattr(self.lua, "_pynvim_{}".format(self.channel_id))
149 def request(self, name, *args, **kwargs):
150 r"""Send an API request or notification to nvim.
152 It is rarely needed to call this function directly, as most API
153 functions have python wrapper functions. The `api` object can
154 be also be used to call API functions as methods:
156 vim.api.err_write('ERROR\n', async_=True)
157 vim.current.buffer.api.get_mark('.')
161 vim.request('nvim_err_write', 'ERROR\n', async_=True)
162 vim.request('nvim_buf_get_mark', vim.current.buffer, '.')
165 Normally a blocking request will be sent. If the `async_` flag is
166 present and True, a asynchronous notification is sent instead. This
167 will never block, and the return value or error is ignored.
169 if (self._session._loop_thread is not None
170 and threading.current_thread() != self._session._loop_thread):
172 msg = ("Request from non-main thread.\n"
173 "Requests from different threads should be wrapped "
174 "with nvim.async_call(cb, ...) \n{}\n"
175 .format('\n'.join(format_stack(None, 5)[:-1])))
177 self.async_call(self._err_cb, msg)
178 raise NvimError("request from non-main thread")
180 decode = kwargs.pop('decode', self._decode)
181 args = walk(self._to_nvim, args)
182 res = self._session.request(name, *args, **kwargs)
183 return walk(self._from_nvim, res, decode=decode)
185 def next_message(self):
186 """Block until a message(request or notification) is available.
188 If any messages were previously enqueued, return the first in queue.
189 If not, run the event loop until one is received.
191 msg = self._session.next_message()
193 return walk(self._from_nvim, msg)
195 def run_loop(self, request_cb, notification_cb,
196 setup_cb=None, err_cb=None):
197 """Run the event loop to receive requests and notifications from Nvim.
199 This should not be called from a plugin running in the host, which
200 already runs the loop and dispatches events to plugins.
203 err_cb = sys.stderr.write
204 self._err_cb = err_cb
206 def filter_request_cb(name, args):
207 name = self._from_nvim(name)
208 args = walk(self._from_nvim, args)
210 result = request_cb(name, args)
212 msg = ("error caught in request handler '{} {}'\n{}\n\n"
213 .format(name, args, format_exc_skip(1)))
216 return walk(self._to_nvim, result)
218 def filter_notification_cb(name, args):
219 name = self._from_nvim(name)
220 args = walk(self._from_nvim, args)
222 notification_cb(name, args)
224 msg = ("error caught in notification handler '{} {}'\n{}\n\n"
225 .format(name, args, format_exc_skip(1)))
229 self._session.run(filter_request_cb, filter_notification_cb, setup_cb)
232 """Stop the event loop being started with `run_loop`."""
236 """Close the nvim session and release its resources."""
237 self._session.close()
240 """Enter nvim session as a context manager."""
243 def __exit__(self, *exc_info):
244 """Exit nvim session as a context manager.
246 Closes the event loop.
250 def with_decode(self, decode=True):
251 """Initialize a new Nvim instance."""
252 return Nvim(self._session, self.channel_id,
253 self.metadata, self.types, decode, self._err_cb)
255 def ui_attach(self, width, height, rgb=None, **kwargs):
256 """Register as a remote UI.
258 After this method is called, the client will receive redraw
264 return self.request('nvim_ui_attach', width, height, options)
267 """Unregister as a remote UI."""
268 return self.request('nvim_ui_detach')
270 def ui_try_resize(self, width, height):
271 """Notify nvim that the client window has resized.
273 If possible, nvim will send a redraw request to resize.
275 return self.request('ui_try_resize', width, height)
277 def subscribe(self, event):
278 """Subscribe to a Nvim event."""
279 return self.request('nvim_subscribe', event)
281 def unsubscribe(self, event):
282 """Unsubscribe to a Nvim event."""
283 return self.request('nvim_unsubscribe', event)
285 def command(self, string, **kwargs):
286 """Execute a single ex command."""
287 return self.request('nvim_command', string, **kwargs)
289 def command_output(self, string):
290 """Execute a single ex command and return the output."""
291 return self.request('nvim_command_output', string)
293 def eval(self, string, **kwargs):
294 """Evaluate a vimscript expression."""
295 return self.request('nvim_eval', string, **kwargs)
297 def call(self, name, *args, **kwargs):
298 """Call a vimscript function."""
299 return self.request('nvim_call_function', name, args, **kwargs)
301 def exec_lua(self, code, *args, **kwargs):
304 Additional parameters are available as `...` inside the lua chunk.
305 Only statements are executed. To evaluate an expression, prefix it
306 with `return`: `return my_function(...)`
308 There is a shorthand syntax to call lua functions with arguments:
311 nvim.lua.mymod.myfunction(data, async_=True)
315 nvim.exec_lua("return func(...)", 1, 2)
316 nvim.exec_lua("mymod.myfunction(...)", data, async_=True)
318 Note that with `async_=True` there is no return value.
320 return self.request('nvim_execute_lua', code, args, **kwargs)
322 def strwidth(self, string):
323 """Return the number of display cells `string` occupies.
325 Tab is counted as one cell.
327 return self.request('nvim_strwidth', string)
329 def list_runtime_paths(self):
330 """Return a list of paths contained in the 'runtimepath' option."""
331 return self.request('nvim_list_runtime_paths')
333 def foreach_rtp(self, cb):
334 """Invoke `cb` for each path in 'runtimepath'.
336 Call the given callable for each path in 'runtimepath' until either
337 callable returns something but None, the exception is raised or there
338 are no longer paths. If stopped in case callable returned non-None,
339 vim.foreach_rtp function returns the value returned by callable.
341 for path in self.request('nvim_list_runtime_paths'):
343 if cb(path) is not None:
348 def chdir(self, dir_path):
349 """Run os.chdir, then all appropriate vim stuff."""
351 return self.request('nvim_set_current_dir', dir_path)
353 def feedkeys(self, keys, options='', escape_csi=True):
354 """Push `keys` to Nvim user input buffer.
356 Options can be a string with the following character flags:
357 - 'm': Remap keys. This is default.
358 - 'n': Do not remap keys.
359 - 't': Handle keys as if typed; otherwise they are handled as if coming
360 from a mapping. This matters for undo, opening folds, etc.
362 return self.request('nvim_feedkeys', keys, options, escape_csi)
364 def input(self, bytes):
365 """Push `bytes` to Nvim low level input buffer.
367 Unlike `feedkeys()`, this uses the lowest level input buffer and the
368 call is not deferred. It returns the number of bytes actually
369 written(which can be less than what was requested if the buffer is
372 return self.request('nvim_input', bytes)
374 def replace_termcodes(self, string, from_part=False, do_lt=True,
376 r"""Replace any terminal code strings by byte sequences.
378 The returned sequences are Nvim's internal representation of keys,
386 The returned sequences can be used as input to `feedkeys`.
388 return self.request('nvim_replace_termcodes', string,
389 from_part, do_lt, special)
391 def out_write(self, msg, **kwargs):
392 r"""Print `msg` as a normal message.
394 The message is buffered (won't display) until linefeed ("\n").
396 return self.request('nvim_out_write', msg, **kwargs)
398 def err_write(self, msg, **kwargs):
399 r"""Print `msg` as an error message.
401 The message is buffered (won't display) until linefeed ("\n").
403 if self._thread_invalid():
404 # special case: if a non-main thread writes to stderr
405 # i.e. due to an uncaught exception, pass it through
406 # without raising an additional exception.
407 self.async_call(self.err_write, msg, **kwargs)
409 return self.request('nvim_err_write', msg, **kwargs)
411 def _thread_invalid(self):
412 return (self._session._loop_thread is not None
413 and threading.current_thread() != self._session._loop_thread)
415 def quit(self, quit_command='qa!'):
416 """Send a quit command to Nvim.
418 By default, the quit command is 'qa!' which will make Nvim quit without
422 self.command(quit_command)
424 # sending a quit command will raise an IOError because the
425 # connection is closed before a response is received. Safe to
429 def new_highlight_source(self):
430 """Return new src_id for use with Buffer.add_highlight."""
431 return self.current.buffer.add_highlight("", 0, src_id=0)
433 def async_call(self, fn, *args, **kwargs):
434 """Schedule `fn` to be called by the event loop soon.
436 This function is thread-safe, and is the only way code not
437 on the main thread could interact with nvim api objects.
439 This function can also be called in a synchronous
440 event handler, just before it returns, to defer execution
441 that shouldn't block neovim.
443 call_point = ''.join(format_stack(None, 5)[:-1])
448 except Exception as err:
449 msg = ("error caught while executing async callback:\n"
450 "{!r}\n{}\n \nthe call was requested at\n{}"
451 .format(err, format_exc_skip(1), call_point))
454 self._session.threadsafe_call(handler)
457 class Buffers(object):
459 """Remote NVim buffers.
461 Currently the interface for interacting with remote NVim buffers is the
462 `nvim_list_bufs` msgpack-rpc function. Most methods fetch the list of
465 Conforms to *python-buffers*.
468 def __init__(self, nvim):
469 """Initialize a Buffers object with Nvim object `nvim`."""
470 self._fetch_buffers = nvim.api.list_bufs
473 """Return the count of buffers."""
474 return len(self._fetch_buffers())
476 def __getitem__(self, number):
477 """Return the Buffer object matching buffer number `number`."""
478 for b in self._fetch_buffers():
479 if b.number == number:
481 raise KeyError(number)
483 def __contains__(self, b):
484 """Return whether Buffer `b` is a known valid buffer."""
485 return isinstance(b, Buffer) and b.valid
488 """Return an iterator over the list of buffers."""
489 return iter(self._fetch_buffers())
492 class CompatibilitySession(object):
494 """Helper class for API compatibility."""
496 def __init__(self, nvim):
497 self.threadsafe_call = nvim.async_call
500 class Current(object):
502 """Helper class for emulating vim.current from python-vim."""
504 def __init__(self, session):
505 self._session = session
510 return self._session.request('nvim_get_current_line')
513 def line(self, line):
514 return self._session.request('nvim_set_current_line', line)
518 return self._session.request('nvim_del_current_line')
522 return self._session.request('nvim_get_current_buf')
525 def buffer(self, buffer):
526 return self._session.request('nvim_set_current_buf', buffer)
530 return self._session.request('nvim_get_current_win')
533 def window(self, window):
534 return self._session.request('nvim_set_current_win', window)
538 return self._session.request('nvim_get_current_tabpage')
541 def tabpage(self, tabpage):
542 return self._session.request('nvim_set_current_tabpage', tabpage)
547 """Helper class for functional vimscript interface."""
549 def __init__(self, nvim):
552 def __getattr__(self, name):
553 return partial(self._nvim.call, name)
556 class LuaFuncs(object):
558 """Wrapper to allow lua functions to be called like python methods."""
560 def __init__(self, nvim, name=""):
564 def __getattr__(self, name):
565 """Return wrapper to named api method."""
566 prefix = self.name + "." if self.name else ""
567 return LuaFuncs(self._nvim, prefix + name)
569 def __call__(self, *args, **kwargs):
570 # first new function after keyword rename, be a bit noisy
571 if 'async' in kwargs:
572 raise ValueError('"async" argument is not allowed. '
573 'Use "async_" instead.')
574 async_ = kwargs.get('async_', False)
575 pattern = "return {}(...)" if not async_ else "{}(...)"
576 code = pattern.format(self.name)
577 return self._nvim.exec_lua(code, *args, **kwargs)