pynvim.api.nvim.Buffers.__init__() - Code Metrics - Inspection of "coveragerc: parallel=1" - neovim/pynvim - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#421)

by Daniel

created 2020-01-10 19:16 UTC

pynvim.api.nvim.Buffers.init() A

↳ Parent: pynvim.api.nvim

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	2
dl	0
loc	3
rs	10
c	0
b	0
f	0
ccs	2
cts	2
cp	1
crap	1

"""Main Nvim interface."""
import os
import sys
import threading
from functools import partial
from traceback import format_stack

from msgpack import ExtType

from .buffer import Buffer
from .common import (NvimError, Remote, RemoteApi, RemoteMap, RemoteSequence,
                     decode_if_bytes, walk)
from .tabpage import Tabpage
from .window import Window
from ..compat import IS_PYTHON3
from ..util import Version, format_exc_skip

__all__ = ('Nvim')


os_chdir = os.chdir

lua_module = """
local a = vim.api
local function update_highlights(buf, src_id, hls, clear_first, clear_end)
  if clear_first ~= nil then
      a.nvim_buf_clear_highlight(buf, src_id, clear_first, clear_end)
  end
  for _,hl in pairs(hls) do
    local group, line, col_start, col_end = unpack(hl)
    if col_start == nil then
      col_start = 0
    end
    if col_end == nil then
      col_end = -1
    end
    a.nvim_buf_add_highlight(buf, src_id, group, line, col_start, col_end)
  end
end

local chid = ...
local mod = {update_highlights=update_highlights}
_G["_pynvim_"..chid] = mod
"""


class Nvim(object):

    """Class that represents a remote Nvim instance.

    This class is main entry point to Nvim remote API, it is a wrapper
    around Session instances.

    The constructor of this class must not be called directly. Instead, the
    `from_session` class method should be used to create the first instance
    from a raw `Session` instance.

    Subsequent instances for the same session can be created by calling the
    `with_decode` instance method to change the decoding behavior or
    `SubClass.from_nvim(nvim)` where `SubClass` is a subclass of `Nvim`, which
    is useful for having multiple `Nvim` objects that behave differently
    without one affecting the other.

    When this library is used on python3.4+, asyncio event loop is guaranteed
    to be used. It is available as the "loop" attribute of this class. Note
    that asyncio callbacks cannot make blocking requests, which includes
    accessing state-dependent attributes. They should instead schedule another
    callback using nvim.async_call, which will not have this restriction.
    """

    @classmethod
    def from_session(cls, session):
        """Create a new Nvim instance for a Session instance.

        This method must be called to create the first Nvim instance, since it
        queries Nvim metadata for type information and sets a SessionHook for
        creating specialized objects from Nvim remote handles.
        """
        session.error_wrapper = lambda e: NvimError(decode_if_bytes(e[1]))
        channel_id, metadata = session.request(b'nvim_get_api_info')

        if IS_PYTHON3:
            # decode all metadata strings for python3
            metadata = walk(decode_if_bytes, metadata)

        types = {
            metadata['types']['Buffer']['id']: Buffer,
            metadata['types']['Window']['id']: Window,
            metadata['types']['Tabpage']['id']: Tabpage,
        }

        return cls(session, channel_id, metadata, types)

    @classmethod
    def from_nvim(cls, nvim):
        """Create a new Nvim instance from an existing instance."""
        return cls(nvim._session, nvim.channel_id, nvim.metadata,
                   nvim.types, nvim._decode, nvim._err_cb)

    def __init__(self, session, channel_id, metadata, types,
                 decode=False, err_cb=None):
        """Initialize a new Nvim instance. This method is module-private."""
        self._session = session
        self.channel_id = channel_id
        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 _from_nvim(self, obj, decode=None):
        if decode is None:
            decode = self._decode
        if type(obj) is ExtType:
            cls = self.types[obj.code]
            return cls(self, (obj.code, obj.data))
        if decode:
            obj = decode_if_bytes(obj, decode)
        return obj

    def _to_nvim(self, obj):
        if isinstance(obj, Remote):
            return ExtType(*obj.code_data)
        return obj

    def _get_lua_private(self):
        if not getattr(self._session, "_has_lua", False):
            self.exec_lua(lua_module, self.channel_id)
            self._session._has_lua = True
        return getattr(self.lua, "_pynvim_{}".format(self.channel_id))

    def request(self, name, *args, **kwargs):
        r"""Send an API request or notification to nvim.

        It is rarely needed to call this function directly, as most API
        functions have python wrapper functions. The `api` object can
        be also be used to call API functions as methods:

            vim.api.err_write('ERROR\n', async_=True)
            vim.current.buffer.api.get_mark('.')

        is equivalent to

            vim.request('nvim_err_write', 'ERROR\n', async_=True)
            vim.request('nvim_buf_get_mark', vim.current.buffer, '.')


        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)

    def next_message(self):
        """Block until a message(request or notification) is available.

        If any messages were previously enqueued, return the first in queue.
        If not, run the event loop until one is received.
        """
        msg = self._session.next_message()
        if msg:
            return walk(self._from_nvim, msg)

    def run_loop(self, request_cb, notification_cb,
                 setup_cb=None, err_cb=None):
        """Run the event loop to receive requests and notifications from Nvim.

        This should not be called from a plugin running in the host, which
        already runs the loop and dispatches events to plugins.
        """
        if err_cb is None:
            err_cb = sys.stderr.write
        self._err_cb = err_cb

        def filter_request_cb(name, args):
            name = self._from_nvim(name)
            args = walk(self._from_nvim, args)
            try:
                result = request_cb(name, args)
            except Exception:
                msg = ("error caught in request handler '{} {}'\n{}\n\n"
                       .format(name, args, format_exc_skip(1)))
                self._err_cb(msg)
                raise
            return walk(self._to_nvim, result)

        def filter_notification_cb(name, args):
            name = self._from_nvim(name)
            args = walk(self._from_nvim, args)
            try:
                notification_cb(name, args)
            except Exception:
                msg = ("error caught in notification handler '{} {}'\n{}\n\n"
                       .format(name, args, format_exc_skip(1)))
                self._err_cb(msg)
                raise

        self._session.run(filter_request_cb, filter_notification_cb, setup_cb)

    def stop_loop(self):
        """Stop the event loop being started with `run_loop`."""
        self._session.stop()

    def close(self):
        """Close the nvim session and release its resources."""
        self._session.close()

    def __enter__(self):
        """Enter nvim session as a context manager."""
        return self

    def __exit__(self, *exc_info):
        """Exit nvim session as a context manager.

        Closes the event loop.
        """
        self.close()

    def with_decode(self, decode=True):
        """Initialize a new Nvim instance."""
        return Nvim(self._session, self.channel_id,
                    self.metadata, self.types, decode, self._err_cb)

    def ui_attach(self, width, height, rgb=None, **kwargs):
        """Register as a remote UI.

        After this method is called, the client will receive redraw
        notifications.
        """
        options = kwargs
        if rgb is not None:
            options['rgb'] = rgb
        return self.request('nvim_ui_attach', width, height, options)

    def ui_detach(self):
        """Unregister as a remote UI."""
        return self.request('nvim_ui_detach')

    def ui_try_resize(self, width, height):
        """Notify nvim that the client window has resized.

        If possible, nvim will send a redraw request to resize.
        """
        return self.request('ui_try_resize', width, height)

    def subscribe(self, event):
        """Subscribe to a Nvim event."""
        return self.request('nvim_subscribe', event)

    def unsubscribe(self, event):
        """Unsubscribe to a Nvim event."""
        return self.request('nvim_unsubscribe', event)

    def command(self, string, **kwargs):
        """Execute a single ex command."""
        return self.request('nvim_command', string, **kwargs)

    def command_output(self, string):
        """Execute a single ex command and return the output."""
        return self.request('nvim_command_output', string)

    def eval(self, string, **kwargs):
        """Evaluate a vimscript expression."""
        return self.request('nvim_eval', string, **kwargs)

    def call(self, name, *args, **kwargs):
        """Call a vimscript function."""
        return self.request('nvim_call_function', name, args, **kwargs)

    def exec_lua(self, code, *args, **kwargs):
        """Execute lua code.

        Additional parameters are available as `...` inside the lua chunk.
        Only statements are executed.  To evaluate an expression, prefix it
        with `return`: `return my_function(...)`

        There is a shorthand syntax to call lua functions with arguments:

            nvim.lua.func(1,2)
            nvim.lua.mymod.myfunction(data, async_=True)

        is equivalent to

            nvim.exec_lua("return func(...)", 1, 2)
            nvim.exec_lua("mymod.myfunction(...)", data, async_=True)

        Note that with `async_=True` there is no return value.
        """
        return self.request('nvim_execute_lua', code, args, **kwargs)

    def strwidth(self, string):
        """Return the number of display cells `string` occupies.

        Tab is counted as one cell.
        """
        return self.request('nvim_strwidth', string)

    def list_runtime_paths(self):
        """Return a list of paths contained in the 'runtimepath' option."""
        return self.request('nvim_list_runtime_paths')

    def foreach_rtp(self, cb):
        """Invoke `cb` for each path in 'runtimepath'.

        Call the given callable for each path in 'runtimepath' until either
        callable returns something but None, the exception is raised or there
        are no longer paths. If stopped in case callable returned non-None,
        vim.foreach_rtp function returns the value returned by callable.
        """
        for path in self.request('nvim_list_runtime_paths'):
            try:
                if cb(path) is not None:
                    break
            except Exception:
                break

    def chdir(self, dir_path):
        """Run os.chdir, then all appropriate vim stuff."""
        os_chdir(dir_path)
        return self.request('nvim_set_current_dir', dir_path)

    def feedkeys(self, keys, options='', escape_csi=True):
        """Push `keys` to Nvim user input buffer.

        Options can be a string with the following character flags:
        - 'm': Remap keys. This is default.
        - 'n': Do not remap keys.
        - 't': Handle keys as if typed; otherwise they are handled as if coming
               from a mapping. This matters for undo, opening folds, etc.
        """
        return self.request('nvim_feedkeys', keys, options, escape_csi)

    def input(self, bytes):
        """Push `bytes` to Nvim low level input buffer.

        Unlike `feedkeys()`, this uses the lowest level input buffer and the
        call is not deferred. It returns the number of bytes actually
        written(which can be less than what was requested if the buffer is
        full).
        """
        return self.request('nvim_input', bytes)

    def replace_termcodes(self, string, from_part=False, do_lt=True,
                          special=True):
        r"""Replace any terminal code strings by byte sequences.

        The returned sequences are Nvim's internal representation of keys,
        for example:

        <esc> -> '\x1b'
        <cr>  -> '\r'
        <c-l> -> '\x0c'
        <up>  -> '\x80ku'

        The returned sequences can be used as input to `feedkeys`.
        """
        return self.request('nvim_replace_termcodes', string,
                            from_part, do_lt, special)

    def out_write(self, msg, **kwargs):
        r"""Print `msg` as a normal message.

        The message is buffered (won't display) until linefeed ("\n").
        """
        return self.request('nvim_out_write', msg, **kwargs)

    def err_write(self, msg, **kwargs):
        r"""Print `msg` as an error message.

        The message is buffered (won't display) until linefeed ("\n").
        """
        if self._thread_invalid():
            # special case: if a non-main thread writes to stderr
            # i.e. due to an uncaught exception, pass it through
            # without raising an additional exception.
            self.async_call(self.err_write, msg, **kwargs)
            return
        return self.request('nvim_err_write', msg, **kwargs)

    def _thread_invalid(self):
        return (self._session._loop_thread is not None
                and threading.current_thread() != self._session._loop_thread)

    def quit(self, quit_command='qa!'):
        """Send a quit command to Nvim.

        By default, the quit command is 'qa!' which will make Nvim quit without
        saving anything.
        """
        try:
            self.command(quit_command)
        except IOError:
            # sending a quit command will raise an IOError because the
            # connection is closed before a response is received. Safe to
            # ignore it.
            pass

    def new_highlight_source(self):
        """Return new src_id for use with Buffer.add_highlight."""
        return self.current.buffer.add_highlight("", 0, src_id=0)

    def async_call(self, fn, *args, **kwargs):
        """Schedule `fn` to be called by the event loop soon.

        This function is thread-safe, and is the only way code not
        on the main thread could interact with nvim api objects.

        This function can also be called in a synchronous
        event handler, just before it returns, to defer execution
        that shouldn't block neovim.
        """
        call_point = ''.join(format_stack(None, 5)[:-1])

        def handler():
            try:
                fn(*args, **kwargs)
            except Exception as err:
                msg = ("error caught while executing async callback:\n"
                       "{!r}\n{}\n \nthe call was requested at\n{}"
                       .format(err, format_exc_skip(1), call_point))
                self._err_cb(msg)
                raise
        self._session.threadsafe_call(handler)


class Buffers(object):

    """Remote NVim buffers.

    Currently the interface for interacting with remote NVim buffers is the
    `nvim_list_bufs` msgpack-rpc function. Most methods fetch the list of
    buffers from NVim.

    Conforms to *python-buffers*.
    """

    def __init__(self, nvim):
        """Initialize a Buffers object with Nvim object `nvim`."""
        self._fetch_buffers = nvim.api.list_bufs

    def __len__(self):
        """Return the count of buffers."""
        return len(self._fetch_buffers())

    def __getitem__(self, number):
        """Return the Buffer object matching buffer number `number`."""
        for b in self._fetch_buffers():
            if b.number == number:
                return b
        raise KeyError(number)

    def __contains__(self, b):
        """Return whether Buffer `b` is a known valid buffer."""
        return isinstance(b, Buffer) and b.valid

    def __iter__(self):
        """Return an iterator over the list of buffers."""
        return iter(self._fetch_buffers())


class CompatibilitySession(object):

    """Helper class for API compatibility."""

    def __init__(self, nvim):
        self.threadsafe_call = nvim.async_call


class Current(object):

    """Helper class for emulating vim.current from python-vim."""

    def __init__(self, session):
        self._session = session
        self.range = None

    @property
    def line(self):
        return self._session.request('nvim_get_current_line')

    @line.setter
    def line(self, line):
        return self._session.request('nvim_set_current_line', line)

    @line.deleter
    def line(self):
        return self._session.request('nvim_del_current_line')

    @property
    def buffer(self):
        return self._session.request('nvim_get_current_buf')

    @buffer.setter
    def buffer(self, buffer):
        return self._session.request('nvim_set_current_buf', buffer)

    @property
    def window(self):
        return self._session.request('nvim_get_current_win')

    @window.setter
    def window(self, window):
        return self._session.request('nvim_set_current_win', window)

    @property
    def tabpage(self):
        return self._session.request('nvim_get_current_tabpage')

    @tabpage.setter
    def tabpage(self, tabpage):
        return self._session.request('nvim_set_current_tabpage', tabpage)


class Funcs(object):

    """Helper class for functional vimscript interface."""

    def __init__(self, nvim):
        self._nvim = nvim

    def __getattr__(self, name):
        return partial(self._nvim.call, name)


class LuaFuncs(object):

    """Wrapper to allow lua functions to be called like python methods."""

    def __init__(self, nvim, name=""):
        self._nvim = nvim
        self.name = name

    def __getattr__(self, name):
        """Return wrapper to named api method."""
        prefix = self.name + "." if self.name else ""
        return LuaFuncs(self._nvim, prefix + name)

    def __call__(self, *args, **kwargs):
        # first new function after keyword rename, be a bit noisy
        if 'async' in kwargs:
            raise ValueError('"async" argument is not allowed. '
                             'Use "async_" instead.')
        async_ = kwargs.get('async_', False)
        pattern = "return {}(...)" if not async_ else "{}(...)"
        code = pattern.format(self.name)
        return self._nvim.exec_lua(code, *args, **kwargs)


1		"""Main Nvim interface."""
2	6	import os
3	6	import sys
4	6	import threading
5	6	from functools import partial
6	6	from traceback import format_stack
7
8	6	from msgpack import ExtType
9
10	6	from .buffer import Buffer
11	6	from .common import (NvimError, Remote, RemoteApi, RemoteMap, RemoteSequence,
12		decode_if_bytes, walk)
13	6	from .tabpage import Tabpage
14	6	from .window import Window
15	6	from ..compat import IS_PYTHON3
16	6	from ..util import Version, format_exc_skip
17
18	6	__all__ = ('Nvim')
19
20
21	6	os_chdir = os.chdir
22
23	6	lua_module = """
24		local a = vim.api
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)
28		end
29		for _,hl in pairs(hls) do
30		local group, line, col_start, col_end = unpack(hl)
31		if col_start == nil then
32		col_start = 0
33		end
34		if col_end == nil then
35		col_end = -1
36		end
37		a.nvim_buf_add_highlight(buf, src_id, group, line, col_start, col_end)
38		end
39		end
40
41		local chid = ...
42		local mod = {update_highlights=update_highlights}
43		_G["_pynvim_"..chid] = mod
44		"""
45
46
47	6	class Nvim(object):
48
49		"""Class that represents a remote Nvim instance.
50
51		This class is main entry point to Nvim remote API, it is a wrapper
52		around Session instances.
53
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.
57
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.
63
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.
69		"""
70
71	6	@classmethod
72		def from_session(cls, session):
73		"""Create a new Nvim instance for a Session instance.
74
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.
78		"""
79	6	session.error_wrapper = lambda e: NvimError(decode_if_bytes(e[1]))
80	6	channel_id, metadata = session.request(b'nvim_get_api_info')
81
82	6	if IS_PYTHON3:
83		# decode all metadata strings for python3
84	4	metadata = walk(decode_if_bytes, metadata)
85
86	6	types = {
87		metadata['types']['Buffer']['id']: Buffer,
88		metadata['types']['Window']['id']: Window,
89		metadata['types']['Tabpage']['id']: Tabpage,
90		}
91
92	6	return cls(session, channel_id, metadata, types)
93
94	6	@classmethod
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)
99
100	6	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	6	self._session = session
104	6	self.channel_id = channel_id
105	6	self.metadata = metadata
106	6	version = metadata.get("version", {"api_level": 0})
107	6	self.version = Version(**version)
108	6	self.types = types
109	6	self.api = RemoteApi(self, 'nvim_')
110	6	self.vars = RemoteMap(self, 'nvim_get_var', 'nvim_set_var', 'nvim_del_var')
111	6	self.vvars = RemoteMap(self, 'nvim_get_vvar', None, None)
112	6	self.options = RemoteMap(self, 'nvim_get_option', 'nvim_set_option')
113	6	self.buffers = Buffers(self)
114	6	self.windows = RemoteSequence(self, 'nvim_list_wins')
115	6	self.tabpages = RemoteSequence(self, 'nvim_list_tabpages')
116	6	self.current = Current(self)
117	6	self.session = CompatibilitySession(self)
118	6	self.funcs = Funcs(self)
119	6	self.lua = LuaFuncs(self)
120	6	self.error = NvimError
121	6	self._decode = decode
122	6	self._err_cb = err_cb
123
124		# only on python3.4+ we expose asyncio
125	6	if IS_PYTHON3:
126	4	self.loop = self._session.loop._loop
127
128	6	def _from_nvim(self, obj, decode=None):
129	6	if decode is None:
130	6	decode = self._decode
131	6	if type(obj) is ExtType:
132	6	cls = self.types[obj.code]
133	6	return cls(self, (obj.code, obj.data))
134	6	if decode:
135	4	obj = decode_if_bytes(obj, decode)
136	6	return obj
137
138	6	def _to_nvim(self, obj):
139	6	if isinstance(obj, Remote):
140	6	return ExtType(*obj.code_data)
141	6	return obj
142
143	6	def _get_lua_private(self):
144	6	if not getattr(self._session, "_has_lua", False):
145	6	self.exec_lua(lua_module, self.channel_id)
146	6	self._session._has_lua = True
147	6	return getattr(self.lua, "_pynvim_{}".format(self.channel_id))
148
149	6	def request(self, name, args, *kwargs):
150		r"""Send an API request or notification to nvim.
151
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:
155
156		vim.api.err_write('ERROR\n', async_=True)
157		vim.current.buffer.api.get_mark('.')
158
159		is equivalent to
160
161		vim.request('nvim_err_write', 'ERROR\n', async_=True)
162		vim.request('nvim_buf_get_mark', vim.current.buffer, '.')
163
164
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.
168		"""
169	6	if (self._session._loop_thread is not None
170		and threading.current_thread() != self._session._loop_thread):
171
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])))
176
177		self.async_call(self._err_cb, msg)
178		raise NvimError("request from non-main thread")
179
180	6	decode = kwargs.pop('decode', self._decode)
181	6	args = walk(self._to_nvim, args)
182	6	res = self._session.request(name, args, *kwargs)
183	6	return walk(self._from_nvim, res, decode=decode)
184
185	6	def next_message(self):
186		"""Block until a message(request or notification) is available.
187
188		If any messages were previously enqueued, return the first in queue.
189		If not, run the event loop until one is received.
190		"""
191	6	msg = self._session.next_message()
192	6	if msg:
193	6	return walk(self._from_nvim, msg)
194
195	6	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.
198
199		This should not be called from a plugin running in the host, which
200		already runs the loop and dispatches events to plugins.
201		"""
202	6	if err_cb is None:
203	6	err_cb = sys.stderr.write
204	6	self._err_cb = err_cb
205
206	6	def filter_request_cb(name, args):
207	6	name = self._from_nvim(name)
208	6	args = walk(self._from_nvim, args)
209	6	try:
210	6	result = request_cb(name, args)
211		except Exception:
212		msg = ("error caught in request handler '{} {}'\n{}\n\n"
213		.format(name, args, format_exc_skip(1)))
214		self._err_cb(msg)
215		raise
216	6	return walk(self._to_nvim, result)
217
218	6	def filter_notification_cb(name, args):
219		name = self._from_nvim(name)
220		args = walk(self._from_nvim, args)
221		try:
222		notification_cb(name, args)
223		except Exception:
224		msg = ("error caught in notification handler '{} {}'\n{}\n\n"
225		.format(name, args, format_exc_skip(1)))
226		self._err_cb(msg)
227		raise
228
229	6	self._session.run(filter_request_cb, filter_notification_cb, setup_cb)
230
231	6	def stop_loop(self):
232		"""Stop the event loop being started with `run_loop`."""
233	6	self._session.stop()
234
235	6	def close(self):
236		"""Close the nvim session and release its resources."""
237		self._session.close()
238
239	6	def __enter__(self):
240		"""Enter nvim session as a context manager."""
241		return self
242
243	6	def __exit__(self, *exc_info):
244		"""Exit nvim session as a context manager.
245
246		Closes the event loop.
247		"""
248		self.close()
249
250	6	def with_decode(self, decode=True):
251		"""Initialize a new Nvim instance."""
252	6	return Nvim(self._session, self.channel_id,
253		self.metadata, self.types, decode, self._err_cb)
254
255	6	def ui_attach(self, width, height, rgb=None, **kwargs):
256		"""Register as a remote UI.
257
258		After this method is called, the client will receive redraw
259		notifications.
260		"""
261		options = kwargs
262		if rgb is not None:
263		options['rgb'] = rgb
264		return self.request('nvim_ui_attach', width, height, options)
265
266	6	def ui_detach(self):
267		"""Unregister as a remote UI."""
268		return self.request('nvim_ui_detach')
269
270	6	def ui_try_resize(self, width, height):
271		"""Notify nvim that the client window has resized.
272
273		If possible, nvim will send a redraw request to resize.
274		"""
275		return self.request('ui_try_resize', width, height)
276
277	6	def subscribe(self, event):
278		"""Subscribe to a Nvim event."""
279	6	return self.request('nvim_subscribe', event)
280
281	6	def unsubscribe(self, event):
282		"""Unsubscribe to a Nvim event."""
283	6	return self.request('nvim_unsubscribe', event)
284
285	6	def command(self, string, **kwargs):
286		"""Execute a single ex command."""
287	6	return self.request('nvim_command', string, **kwargs)
288
289	6	def command_output(self, string):
290		"""Execute a single ex command and return the output."""
291	6	return self.request('nvim_command_output', string)
292
293	6	def eval(self, string, **kwargs):
294		"""Evaluate a vimscript expression."""
295	6	return self.request('nvim_eval', string, **kwargs)
296
297	6	def call(self, name, args, *kwargs):
298		"""Call a vimscript function."""
299	6	return self.request('nvim_call_function', name, args, **kwargs)
300
301	6	def exec_lua(self, code, args, *kwargs):
302		"""Execute lua code.
303
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(...)`
307
308		There is a shorthand syntax to call lua functions with arguments:
309
310		nvim.lua.func(1,2)
311		nvim.lua.mymod.myfunction(data, async_=True)
312
313		is equivalent to
314
315		nvim.exec_lua("return func(...)", 1, 2)
316		nvim.exec_lua("mymod.myfunction(...)", data, async_=True)
317
318		Note that with `async_=True` there is no return value.
319		"""
320	6	return self.request('nvim_execute_lua', code, args, **kwargs)
321
322	6	def strwidth(self, string):
323		"""Return the number of display cells `string` occupies.
324
325		Tab is counted as one cell.
326		"""
327	6	return self.request('nvim_strwidth', string)
328
329	6	def list_runtime_paths(self):
330		"""Return a list of paths contained in the 'runtimepath' option."""
331	1	return self.request('nvim_list_runtime_paths')
332
333	6	def foreach_rtp(self, cb):
334		"""Invoke `cb` for each path in 'runtimepath'.
335
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.
340		"""
341		for path in self.request('nvim_list_runtime_paths'):
342		try:
343		if cb(path) is not None:
344		break
345		except Exception:
346		break
347
348	6	def chdir(self, dir_path):
349		"""Run os.chdir, then all appropriate vim stuff."""
350	6	os_chdir(dir_path)
351	6	return self.request('nvim_set_current_dir', dir_path)
352
353	6	def feedkeys(self, keys, options='', escape_csi=True):
354		"""Push `keys` to Nvim user input buffer.
355
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.
361		"""
362		return self.request('nvim_feedkeys', keys, options, escape_csi)
363
364	6	def input(self, bytes):
365		"""Push `bytes` to Nvim low level input buffer.
366
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
370		full).
371		"""
372	6	return self.request('nvim_input', bytes)
373
374	6	def replace_termcodes(self, string, from_part=False, do_lt=True,
375		special=True):
376		r"""Replace any terminal code strings by byte sequences.
377
378		The returned sequences are Nvim's internal representation of keys,
379		for example:
380
381		<esc> -> '\x1b'
382		<cr> -> '\r'
383		<c-l> -> '\x0c'
384		<up> -> '\x80ku'
385
386		The returned sequences can be used as input to `feedkeys`.
387		"""
388		return self.request('nvim_replace_termcodes', string,
389		from_part, do_lt, special)
390
391	6	def out_write(self, msg, **kwargs):
392		r"""Print `msg` as a normal message.
393
394		The message is buffered (won't display) until linefeed ("\n").
395		"""
396		return self.request('nvim_out_write', msg, **kwargs)
397
398	6	def err_write(self, msg, **kwargs):
399		r"""Print `msg` as an error message.
400
401		The message is buffered (won't display) until linefeed ("\n").
402		"""
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)
408		return
409		return self.request('nvim_err_write', msg, **kwargs)
410
411	6	def _thread_invalid(self):
412		return (self._session._loop_thread is not None
413		and threading.current_thread() != self._session._loop_thread)
414
415	6	def quit(self, quit_command='qa!'):
416		"""Send a quit command to Nvim.
417
418		By default, the quit command is 'qa!' which will make Nvim quit without
419		saving anything.
420		"""
421	1	try:
422	1	self.command(quit_command)
423		except IOError:
424		# sending a quit command will raise an IOError because the
425		# connection is closed before a response is received. Safe to
426		# ignore it.
427	1	pass
428
429	6	def new_highlight_source(self):
430		"""Return new src_id for use with Buffer.add_highlight."""
431	6	return self.current.buffer.add_highlight("", 0, src_id=0)
432
433	6	def async_call(self, fn, args, *kwargs):
434		"""Schedule `fn` to be called by the event loop soon.
435
436		This function is thread-safe, and is the only way code not
437		on the main thread could interact with nvim api objects.
438
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.
442		"""
443	6	call_point = ''.join(format_stack(None, 5)[:-1])
444
445	6	def handler():
446	6	try:
447	6	fn(args, *kwargs)
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))
452		self._err_cb(msg)
453		raise
454	6	self._session.threadsafe_call(handler)
455
456
457	6	class Buffers(object):
458
459		"""Remote NVim buffers.
460
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
463		buffers from NVim.
464
465		Conforms to python-buffers.
466		"""
467
468	6	def __init__(self, nvim):
469		"""Initialize a Buffers object with Nvim object `nvim`."""
470	6	self._fetch_buffers = nvim.api.list_bufs
471
472	6	def __len__(self):
473		"""Return the count of buffers."""
474	6	return len(self._fetch_buffers())
475
476	6	def __getitem__(self, number):
477		"""Return the Buffer object matching buffer number `number`."""
478	6	for b in self._fetch_buffers():
479	6	if b.number == number:
480	6	return b
481		raise KeyError(number)
482
483	6	def __contains__(self, b):
484		"""Return whether Buffer `b` is a known valid buffer."""
485	6	return isinstance(b, Buffer) and b.valid
486
487	6	def __iter__(self):
488		"""Return an iterator over the list of buffers."""
489	6	return iter(self._fetch_buffers())
490
491
492	6	class CompatibilitySession(object):
493
494		"""Helper class for API compatibility."""
495
496	6	def __init__(self, nvim):
497	6	self.threadsafe_call = nvim.async_call
498
499
500	6	class Current(object):
501
502		"""Helper class for emulating vim.current from python-vim."""
503
504	6	def __init__(self, session):
505	6	self._session = session
506	6	self.range = None
507
508	6	@property
509		def line(self):
510	6	return self._session.request('nvim_get_current_line')
511
512	6	@line.setter
513		def line(self, line):
514	6	return self._session.request('nvim_set_current_line', line)
515
516	6	@line.deleter
517		def line(self):
518	6	return self._session.request('nvim_del_current_line')
519
520	6	@property
521		def buffer(self):
522	6	return self._session.request('nvim_get_current_buf')
523
524	6	@buffer.setter
525		def buffer(self, buffer):
526	6	return self._session.request('nvim_set_current_buf', buffer)
527
528	6	@property
529		def window(self):
530	6	return self._session.request('nvim_get_current_win')
531
532	6	@window.setter
533		def window(self, window):
534	6	return self._session.request('nvim_set_current_win', window)
535
536	6	@property
537		def tabpage(self):
538	6	return self._session.request('nvim_get_current_tabpage')
539
540	6	@tabpage.setter
541		def tabpage(self, tabpage):
542	6	return self._session.request('nvim_set_current_tabpage', tabpage)
543
544
545	6	class Funcs(object):
546
547		"""Helper class for functional vimscript interface."""
548
549	6	def __init__(self, nvim):
550	6	self._nvim = nvim
551
552	6	def __getattr__(self, name):
553	6	return partial(self._nvim.call, name)
554
555
556	6	class LuaFuncs(object):
557
558		"""Wrapper to allow lua functions to be called like python methods."""
559
560	6	def __init__(self, nvim, name=""):
561	6	self._nvim = nvim
562	6	self.name = name
563
564	6	def __getattr__(self, name):
565		"""Return wrapper to named api method."""
566	6	prefix = self.name + "." if self.name else ""
567	6	return LuaFuncs(self._nvim, prefix + name)
568
569	6	def __call__(self, args, *kwargs):
570		# first new function after keyword rename, be a bit noisy
571	6	if 'async' in kwargs:
572		raise ValueError('"async" argument is not allowed. '
573		'Use "async_" instead.')
574	6	async_ = kwargs.get('async_', False)
575	6	pattern = "return {}(...)" if not async_ else "{}(...)"
576	6	code = pattern.format(self.name)
577		return self._nvim.exec_lua(code, args, *kwargs)
578

neovim / pynvim

Pull Request — master (#421)

pynvim.api.nvim.Buffers.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

pynvim.api.nvim.Buffers.init() A