Nvim.out_write() - Code Metrics - Inspection of "[WIP] Expose asyncio to python3 rplugins, removing..." - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#294)

by Björn

created 2018-03-01 18:01 UTC

Nvim.out_write() A

↳ Parent: Nvim

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.125

Importance

Changes

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

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

from msgpack import ExtType
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3

from .buffer import Buffer
from .common import (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


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(e[1])
        channel_id, metadata = session.request(b'vim_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,
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
                   nvim.types, nvim._decode, nvim._err_cb)
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent

    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')
        self.vvars = RemoteMap(self, 'nvim_get_vvar', 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.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
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent

    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):
class Foo:
    def some_method(self, x, y):
        return x + y;
        if isinstance(obj, Remote):
            return ExtType(*obj.code_data)
        return obj

    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.
        """
        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):
        """Register as a remote UI.

        After this method is called, the client will receive redraw
        notifications.
        """
        return self.request('ui_attach', width, height, rgb)

    def ui_detach(self):
        """Unregister as a remote UI."""
        return self.request('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 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):
        """Print `msg` as a normal message."""
        return self.request('nvim_out_write', msg, **kwargs)

    def err_write(self, msg, **kwargs):
        """Print `msg` as an error message."""
        return self.request('nvim_err_write', msg, **kwargs)

    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:
try:
    raises_exception()
except:  # Could be removed
    pass
            # 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)

    @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 NvimError(Exception):
    pass


Pull Request — master (#294)

Nvim.out_write() A

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing init.py files

1		"""Main Nvim interface."""
2	5	import os
3	5	import sys
4	5	from functools import partial
5		from traceback import format_stack
6	5
7		from msgpack import ExtType
		0 ignored issues – show Configuration introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The import `msgpack` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
8	5
9		from .buffer import Buffer
10	5	from .common import (Remote, RemoteApi, RemoteMap, RemoteSequence,
11	5	decode_if_bytes, walk)
12		from .tabpage import Tabpage
13	5	from .window import Window
14	5	from ..compat import IS_PYTHON3
15	5	from ..util import Version, format_exc_skip
16	5
17		__all__ = ('Nvim')
18	5
19
20		os_chdir = os.chdir
21	5
22
23		class Nvim(object):
24	5
25		"""Class that represents a remote Nvim instance.
26
27		This class is main entry point to Nvim remote API, it is a wrapper
28		around Session instances.
29
30		The constructor of this class must not be called directly. Instead, the
31		`from_session` class method should be used to create the first instance
32		from a raw `Session` instance.
33
34		Subsequent instances for the same session can be created by calling the
35		`with_decode` instance method to change the decoding behavior or
36		`SubClass.from_nvim(nvim)` where `SubClass` is a subclass of `Nvim`, which
37		is useful for having multiple `Nvim` objects that behave differently
38		without one affecting the other.
39
40		When this library is used on python3.4+, asyncio event loop is guaranteed
41		to be used. It is available as the "loop" attribute of this class. Note
42	5	that asyncio callbacks cannot make blocking requests, which includes
43		accessing state-dependent attributes. They should instead schedule another
44		callback using nvim.async_call, which will not have this restriction.
45		"""
46
47		@classmethod
48		def from_session(cls, session):
49		"""Create a new Nvim instance for a Session instance.
50	5
51	5	This method must be called to create the first Nvim instance, since it
52		queries Nvim metadata for type information and sets a SessionHook for
53	5	creating specialized objects from Nvim remote handles.
54		"""
55	5	session.error_wrapper = lambda e: NvimError(e[1])
56		channel_id, metadata = session.request(b'vim_get_api_info')
57	5
58		if IS_PYTHON3:
59		# decode all metadata strings for python3
60		metadata = walk(decode_if_bytes, metadata)
61
62		types = {
63	5	metadata['types']['Buffer']['id']: Buffer,
64		metadata['types']['Window']['id']: Window,
65	5	metadata['types']['Tabpage']['id']: Tabpage,
66		}
67
68		return cls(session, channel_id, metadata, types)
69
70		@classmethod
71	5	def from_nvim(cls, nvim):
72		"""Create a new Nvim instance from an existing instance."""
73		return cls(nvim._session, nvim.channel_id, nvim.metadata,
		0 ignored issues – show Coding Style Best Practice introduced 2016-03-20 16:11 UTC by Report Bug Copy Issue Report It seems like `_session` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
74	5	nvim.types, nvim._decode, nvim._err_cb)
		0 ignored issues – show Coding Style Best Practice introduced 2016-03-24 17:40 UTC by Report Bug Copy Issue Report It seems like `_decode` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history... Coding Style Best Practice introduced 2016-03-20 11:50 UTC by Report Bug Copy Issue Report It seems like `_err_cb` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
75	5
76	5	def __init__(self, session, channel_id, metadata, types,
77	5	decode=False, err_cb=None):
78	5	"""Initialize a new Nvim instance. This method is module-private."""
79	5	self._session = session
80	5	self.channel_id = channel_id
81	5	self.metadata = metadata
82	5	version = metadata.get("version", {"api_level": 0})
83	5	self.version = Version(**version)
84	5	self.types = types
85	5	self.api = RemoteApi(self, 'nvim_')
86	5	self.vars = RemoteMap(self, 'nvim_get_var', 'nvim_set_var')
87	5	self.vvars = RemoteMap(self, 'nvim_get_vvar', None)
88	5	self.options = RemoteMap(self, 'nvim_get_option', 'nvim_set_option')
89	5	self.buffers = Buffers(self)
90	5	self.windows = RemoteSequence(self, 'nvim_list_wins')
91	5	self.tabpages = RemoteSequence(self, 'nvim_list_tabpages')
92	5	self.current = Current(self)
93		self.session = CompatibilitySession(self)
94	5	self.funcs = Funcs(self)
95	5	self.error = NvimError
96	5	self._decode = decode
97	5	self._err_cb = err_cb
98	5
99	5	# only on python3.4+ we expose asyncio
100	5	if IS_PYTHON3:
101	3	self.loop = self._session.loop._loop
		0 ignored issues – show Coding Style Best Practice introduced 2018-01-23 10:19 UTC by Report Bug Copy Issue Report It seems like `_loop` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
102	5
103		def _from_nvim(self, obj, decode=None):
104	5	if decode is None:
105	5	decode = self._decode
106	5	if type(obj) is ExtType:
107	5	cls = self.types[obj.code]
108		return cls(self, (obj.code, obj.data))
109	5	if decode:
110		obj = decode_if_bytes(obj, decode)
111		return obj
112
113		def _to_nvim(self, obj):
		0 ignored issues – show Coding Style introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report This method could be written as a function/class method. If a method does not access any attributes of the class, it could also be implemented as a function or static method. This can help improve readability. For example class Foo: def some_method(self, x, y): return x + y; could be written as class Foo: @classmethod def some_method(cls, x, y): return x + y; Loading history...
114		if isinstance(obj, Remote):
115		return ExtType(*obj.code_data)
116		return obj
117
118		def request(self, name, args, *kwargs):
119		r"""Send an API request or notification to nvim.
120
121		It is rarely needed to call this function directly, as most API
122		functions have python wrapper functions. The `api` object can
123		be also be used to call API functions as methods:
124
125		vim.api.err_write('ERROR\n', async=True)
126		vim.current.buffer.api.get_mark('.')
127
128		is equivalent to
129	5
130	5	vim.request('nvim_err_write', 'ERROR\n', async=True)
131	5	vim.request('nvim_buf_get_mark', vim.current.buffer, '.')
132	5
133
134	5	Normally a blocking request will be sent. If the `async` flag is
135		present and True, a asynchronous notification is sent instead. This
136		will never block, and the return value or error is ignored.
137		"""
138		decode = kwargs.pop('decode', self._decode)
139		args = walk(self._to_nvim, args)
140	5	res = self._session.request(name, args, *kwargs)
141	5	return walk(self._from_nvim, res, decode=decode)
142	5
143		def next_message(self):
144	5	"""Block until a message(request or notification) is available.
145
146		If any messages were previously enqueued, return the first in queue.
147		If not, run the event loop until one is received.
148		"""
149		msg = self._session.next_message()
150		if msg:
151	5	return walk(self._from_nvim, msg)
152	5
153	5	def run_loop(self, request_cb, notification_cb,
154		setup_cb=None, err_cb=None):
155	5	"""Run the event loop to receive requests and notifications from Nvim.
156	5
157	5	This should not be called from a plugin running in the host, which
158	5	already runs the loop and dispatches events to plugins.
159	5	"""
160		if err_cb is None:
161		err_cb = sys.stderr.write
162		self._err_cb = err_cb
163
164		def filter_request_cb(name, args):
165	5	name = self._from_nvim(name)
166		args = walk(self._from_nvim, args)
167	5	try:
168		result = request_cb(name, args)
169		except Exception:
170		msg = ("error caught in request handler '{} {}'\n{}\n\n"
171		.format(name, args, format_exc_skip(1)))
172		self._err_cb(msg)
173		raise
174		return walk(self._to_nvim, result)
175
176		def filter_notification_cb(name, args):
177		name = self._from_nvim(name)
178	5	args = walk(self._from_nvim, args)
179		try:
180	5	notification_cb(name, args)
181		except Exception:
182	5	msg = ("error caught in notification handler '{} {}'\n{}\n\n"
183		.format(name, args, format_exc_skip(1)))
184	5	self._err_cb(msg)
185		raise
186	5
187		self._session.run(filter_request_cb, filter_notification_cb, setup_cb)
188
189	5	def stop_loop(self):
190		"""Stop the event loop being started with `run_loop`."""
191		self._session.stop()
192
193		def close(self):
194		"""Close the nvim session and release its resources."""
195		self._session.close()
196
197	5	def __enter__(self):
198		"""Enter nvim session as a context manager."""
199		return self
200
201	5	def __exit__(self, *exc_info):
202		"""Exit nvim session as a context manager.
203
204		Closes the event loop.
205		"""
206		self.close()
207
208	5	def with_decode(self, decode=True):
209		"""Initialize a new Nvim instance."""
210	5	return Nvim(self._session, self.channel_id,
211		self.metadata, self.types, decode, self._err_cb)
212	5
213		def ui_attach(self, width, height, rgb):
214	5	"""Register as a remote UI.
215
216	5	After this method is called, the client will receive redraw
217		notifications.
218	5	"""
219		return self.request('ui_attach', width, height, rgb)
220	5
221		def ui_detach(self):
222		"""Unregister as a remote UI."""
223		return self.request('ui_detach')
224	5
225		def ui_try_resize(self, width, height):
226	5	"""Notify nvim that the client window has resized.
227
228	5	If possible, nvim will send a redraw request to resize.
229		"""
230	5	return self.request('ui_try_resize', width, height)
231
232	5	def subscribe(self, event):
233		"""Subscribe to a Nvim event."""
234		return self.request('nvim_subscribe', event)
235
236		def unsubscribe(self, event):
237	5	"""Unsubscribe to a Nvim event."""
238		return self.request('nvim_unsubscribe', event)
239	5
240		def command(self, string, **kwargs):
241		"""Execute a single ex command."""
242		return self.request('nvim_command', string, **kwargs)
243	5
244		def command_output(self, string):
245		"""Execute a single ex command and return the output."""
246		return self.request('nvim_command_output', string)
247
248		def eval(self, string, **kwargs):
249		"""Evaluate a vimscript expression."""
250		return self.request('nvim_eval', string, **kwargs)
251
252		def call(self, name, args, *kwargs):
253		"""Call a vimscript function."""
254		return self.request('nvim_call_function', name, args, **kwargs)
255
256		def strwidth(self, string):
257		"""Return the number of display cells `string` occupies.
258	5
259		Tab is counted as one cell.
260	5	"""
261	5	return self.request('nvim_strwidth', string)
262
263	5	def list_runtime_paths(self):
264		"""Return a list of paths contained in the 'runtimepath' option."""
265		return self.request('nvim_list_runtime_paths')
266
267		def foreach_rtp(self, cb):
268		"""Invoke `cb` for each path in 'runtimepath'.
269
270		Call the given callable for each path in 'runtimepath' until either
271		callable returns something but None, the exception is raised or there
272		are no longer paths. If stopped in case callable returned non-None,
273		vim.foreach_rtp function returns the value returned by callable.
274	5	"""
275		for path in self.request('nvim_list_runtime_paths'):
276		try:
277		if cb(path) is not None:
278		break
279		except Exception:
		0 ignored issues – show Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
280		break
281
282	5	def chdir(self, dir_path):
283		"""Run os.chdir, then all appropriate vim stuff."""
284	5	os_chdir(dir_path)
285		return self.request('nvim_set_current_dir', dir_path)
286
287		def feedkeys(self, keys, options='', escape_csi=True):
288		"""Push `keys` to Nvim user input buffer.
289
290		Options can be a string with the following character flags:
291		- 'm': Remap keys. This is default.
292		- 'n': Do not remap keys.
293		- 't': Handle keys as if typed; otherwise they are handled as if coming
294		from a mapping. This matters for undo, opening folds, etc.
295		"""
296		return self.request('nvim_feedkeys', keys, options, escape_csi)
297
298		def input(self, bytes):
		0 ignored issues – show Bug Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `bytes`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
299		"""Push `bytes` to Nvim low level input buffer.
300
301	5	Unlike `feedkeys()`, this uses the lowest level input buffer and the
302		call is not deferred. It returns the number of bytes actually
303		written(which can be less than what was requested if the buffer is
304		full).
305	5	"""
306		return self.request('nvim_input', bytes)
307	2
308		def replace_termcodes(self, string, from_part=False, do_lt=True,
309	5	special=True):
310		r"""Replace any terminal code strings by byte sequences.
311
312		The returned sequences are Nvim's internal representation of keys,
313		for example:
314
315		<esc> -> '\x1b'
316		<cr> -> '\r'
317		<c-l> -> '\x0c'
318		<up> -> '\x80ku'
319
320		The returned sequences can be used as input to `feedkeys`.
321		"""
322		return self.request('nvim_replace_termcodes', string,
323	5	from_part, do_lt, special)
324
325		def out_write(self, msg, **kwargs):
326		"""Print `msg` as a normal message."""
327	5	return self.request('nvim_out_write', msg, **kwargs)
328
329		def err_write(self, msg, **kwargs):
330		"""Print `msg` as an error message."""
331		return self.request('nvim_err_write', msg, **kwargs)
332
333		def quit(self, quit_command='qa!'):
334		"""Send a quit command to Nvim.
335
336		By default, the quit command is 'qa!' which will make Nvim quit without
337	5	saving anything.
338		"""
339	5	try:
340	5	self.command(quit_command)
341	5	except IOError:
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report This except handler seems to be unused and could be removed. Except handlers which only contain `pass` and do not have an `else` clause can usually simply be removed: try: raises_exception() except: # Could be removed pass Loading history...
342		# sending a quit command will raise an IOError because the
343	1	# connection is closed before a response is received. Safe to
344		# ignore it.
345		pass
346	1
347		def new_highlight_source(self):
348	5	"""Return new src_id for use with Buffer.add_highlight."""
349		return self.current.buffer.add_highlight("", 0, src_id=0)
350
351	5	def async_call(self, fn, args, *kwargs):
352		"""Schedule `fn` to be called by the event loop soon.
353
354		This function is thread-safe, and is the only way code not
355		on the main thread could interact with nvim api objects.
356
357		This function can also be called in a synchronous
358		event handler, just before it returns, to defer execution
359		that shouldn't block neovim.
360		"""
361		call_point = ''.join(format_stack(None, 5)[:-1])
362	5
363		def handler():
364	5	try:
365		fn(args, *kwargs)
366	5	except Exception as err:
367		msg = ("error caught while executing async callback:\n"
368	5	"{!r}\n{}\n \nthe call was requested at\n{}"
369		.format(err, format_exc_skip(1), call_point))
370	5	self._err_cb(msg)
371		raise
372	5	self._session.threadsafe_call(handler)
373	5
374	5
375		class Buffers(object):
376
377	5	"""Remote NVim buffers.
378
379	5	Currently the interface for interacting with remote NVim buffers is the
380		`nvim_list_bufs` msgpack-rpc function. Most methods fetch the list of
381	5	buffers from NVim.
382
383	5	Conforms to python-buffers.
384		"""
385
386	5	def __init__(self, nvim):
387		"""Initialize a Buffers object with Nvim object `nvim`."""
388		self._fetch_buffers = nvim.api.list_bufs
389
390	5	def __len__(self):
391	5	"""Return the count of buffers."""
392		return len(self._fetch_buffers())
393
394	5	def __getitem__(self, number):
395		"""Return the Buffer object matching buffer number `number`."""
396		for b in self._fetch_buffers():
397		if b.number == number:
398	5	return b
399	5	raise KeyError(number)
400	5
401		def __contains__(self, b):
402	5	"""Return whether Buffer `b` is a known valid buffer."""
403		return isinstance(b, Buffer) and b.valid
404	5
405		def __iter__(self):
406	5	"""Return an iterator over the list of buffers."""
407		return iter(self._fetch_buffers())
408	5
409
410	5	class CompatibilitySession(object):
411
412	5	"""Helper class for API compatibility."""
413
414	5	def __init__(self, nvim):
415		self.threadsafe_call = nvim.async_call
416	5
417
418	5	class Current(object):
419
420	5	"""Helper class for emulating vim.current from python-vim."""
421
422	5	def __init__(self, session):
423		self._session = session
424	5	self.range = None
425
426	5	@property
427		def line(self):
428	5	return self._session.request('nvim_get_current_line')
429
430	5	@line.setter
431		def line(self, line):
432	5	return self._session.request('nvim_set_current_line', line)
433
434		@property
435	5	def buffer(self):
436		return self._session.request('nvim_get_current_buf')
437
438		@buffer.setter
439	5	def buffer(self, buffer):
440	5	return self._session.request('nvim_set_current_buf', buffer)
441
442	5	@property
443	5	def window(self):
444		return self._session.request('nvim_get_current_win')
445
446	5	@window.setter
447	5	def window(self, window):
448		return self._session.request('nvim_set_current_win', window)
449
450		@property
451		def tabpage(self):
452		return self._session.request('nvim_get_current_tabpage')
453
454		@tabpage.setter
455		def tabpage(self, tabpage):
456		return self._session.request('nvim_set_current_tabpage', tabpage)
457
458
459		class Funcs(object):
460
461		"""Helper class for functional vimscript interface."""
462
463		def __init__(self, nvim):
464		self._nvim = nvim
465
466		def __getattr__(self, name):
467		return partial(self._nvim.call, name)
468
469
470		class NvimError(Exception):
471		pass
472

neovim / python-client

Pull Request — master (#294)

Nvim.out_write() A

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files