neovim.api.Nvim.input() - Code Metrics - Inspection of "better handling of exceptions in async contexts" - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#169)

by Björn

created 2016-02-28 12:52 UTC

neovim.api.Nvim.input() A

↳ Parent: neovim.api.Nvim

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Metric	Value
cc	1
dl	0
loc	9
ccs	2
cts	2
cp	1
crap	1
rs	9.6666

"""Main Nvim interface."""
import functools
import os

from traceback import format_exc, 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 (DecodeHook, Remote, RemoteMap, RemoteSequence,
                     SessionFilter, SessionHook, walk)
from .tabpage import Tabpage
from .window import Window
from ..compat import IS_PYTHON3


__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 thin 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_hook` instance method and passing a SessionHook instance. This can
    be useful to have multiple `Nvim` objects that behave differently without
    one affecting the other.
    """

    @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:
            hook = DecodeHook()
            # decode all metadata strings for python3
            metadata = walk(hook.from_nvim, metadata, None, None, None)

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

        return cls(session, channel_id, metadata).with_hook(ExtHook(types))

    def __init__(self, session, channel_id, metadata):
        """Initialize a new Nvim instance. This method is module-private."""
        self._session = session
        self.channel_id = channel_id
        self.metadata = metadata
        self.vars = RemoteMap(session, 'vim_get_var', 'vim_set_var')
        self.vvars = RemoteMap(session, 'vim_get_vvar', None)
        self.options = RemoteMap(session, 'vim_get_option', 'vim_set_option')
        self.buffers = RemoteSequence(session, 'vim_get_buffers')
        self.windows = RemoteSequence(session, 'vim_get_windows')
        self.tabpages = RemoteSequence(session, 'vim_get_tabpages')
        self.current = Current(session)
        self.funcs = Funcs(self)
        self.error = NvimError

    def with_hook(self, hook):
        """Initialize a new Nvim instance."""
        return Nvim(SessionFilter(self.session, hook), self.channel_id,
                    self.metadata)

    @property
    def session(self):
        """Return the Session or SessionFilter for a Nvim instance."""
        return self._session

    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._session.request('ui_attach', width, height, rgb)

    def ui_detach(self):
        """Unregister as a remote UI."""
        return self._session.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._session.request('ui_try_resize', width, height)

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

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

    def command(self, string, async=False):
        """Execute a single ex command."""
        return self._session.request('vim_command', string, async=async)

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

    def eval(self, string, async=False):
        """Evaluate a vimscript expression."""
        return self._session.request('vim_eval', string, async=async)

    def call(self, name, *args, **kwargs):
        """Call a vimscript function."""
        for k in kwargs:
            if k != "async":
                raise TypeError(
                    "call() got an unexpected keyword argument '{}'".format(k))
        return self._session.request('vim_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._session.request('vim_strwidth', string)

    def list_runtime_paths(self):
        """Return a list of paths contained in the 'runtimepath' option."""
        return self._session.request('vim_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._session.request('vim_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._session.request('vim_change_directory', 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._session.request('vim_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._session.request('vim_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._session.request('vim_replace_termcodes', string,
                                     from_part, do_lt, special)

    def out_write(self, msg):
        """Print `msg` as a normal message."""
        return self._session.request('vim_out_write', msg)

    def err_write(self, msg, async=False):
        """Print `msg` as an error message."""
        return self._session.request('vim_err_write', msg, async=async)

    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(5), call_point))
                self.err_write(msg, async=True)
                raise
        self._session.threadsafe_call(handler)


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('vim_get_current_line')

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

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

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

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

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

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

    @tabpage.setter
    def tabpage(self, tabpage):
        return self._session.request('vim_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 functools.partial(self._nvim.call, name)


class ExtHook(SessionHook):
    def __init__(self, types):
        self.types = types
        super(ExtHook, self).__init__(from_nvim=self.from_ext,
                                      to_nvim=self.to_ext)

    def from_ext(self, obj, session, method, kind):

        if type(obj) is ExtType:
            cls = self.types[obj.code]
            return cls(session, (obj.code, obj.data))
        return obj

    def to_ext(self, obj, session, method, kind):
class Foo:
    def some_method(self, x, y):
        return x + y;
        if isinstance(obj, Remote):
            return ExtType(*obj.code_data)
        return obj


class NvimError(Exception):
    pass


Pull Request — master (#169)

neovim.api.Nvim.input() A

Complexity

Size

Duplication

Code Coverage

1. Missing Dependencies

2. Missing init.py files

1		"""Main Nvim interface."""
2	6	import functools
3	6	import os
4
5	6	from traceback import format_exc, format_stack
6
7	6	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
9	6	from .buffer import Buffer
10	6	from .common import (DecodeHook, Remote, RemoteMap, RemoteSequence,
11		SessionFilter, SessionHook, walk)
12	6	from .tabpage import Tabpage
13	6	from .window import Window
14	6	from ..compat import IS_PYTHON3
15
16
17	6	__all__ = ('Nvim')
18
19
20	6	os_chdir = os.chdir
21
22
23	6	class Nvim(object):
24
25		"""Class that represents a remote Nvim instance.
26
27		This class is main entry point to Nvim remote API, it is a thin 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_hook` instance method and passing a SessionHook instance. This can
36		be useful to have multiple `Nvim` objects that behave differently without
37		one affecting the other.
38		"""
39
40	6	@classmethod
41		def from_session(cls, session):
42		"""Create a new Nvim instance for a Session instance.
43
44		This method must be called to create the first Nvim instance, since it
45		queries Nvim metadata for type information and sets a SessionHook for
46		creating specialized objects from Nvim remote handles.
47		"""
48	6	session.error_wrapper = lambda e: NvimError(e[1])
49	6	channel_id, metadata = session.request(b'vim_get_api_info')
50
51	6	if IS_PYTHON3:
52	6	hook = DecodeHook()
53		# decode all metadata strings for python3
54	6	metadata = walk(hook.from_nvim, metadata, None, None, None)
55
56	6	types = {
57		metadata['types']['Buffer']['id']: Buffer,
58		metadata['types']['Window']['id']: Window,
59		metadata['types']['Tabpage']['id']: Tabpage,
60		}
61
62	6	return cls(session, channel_id, metadata).with_hook(ExtHook(types))
63
64	6	def __init__(self, session, channel_id, metadata):
65		"""Initialize a new Nvim instance. This method is module-private."""
66	6	self._session = session
67	6	self.channel_id = channel_id
68	6	self.metadata = metadata
69	6	self.vars = RemoteMap(session, 'vim_get_var', 'vim_set_var')
70	6	self.vvars = RemoteMap(session, 'vim_get_vvar', None)
71	6	self.options = RemoteMap(session, 'vim_get_option', 'vim_set_option')
72	6	self.buffers = RemoteSequence(session, 'vim_get_buffers')
73	6	self.windows = RemoteSequence(session, 'vim_get_windows')
74	6	self.tabpages = RemoteSequence(session, 'vim_get_tabpages')
75	6	self.current = Current(session)
76	6	self.funcs = Funcs(self)
77	6	self.error = NvimError
78
79	6	def with_hook(self, hook):
80		"""Initialize a new Nvim instance."""
81	6	return Nvim(SessionFilter(self.session, hook), self.channel_id,
82		self.metadata)
83
84	6	@property
85		def session(self):
86		"""Return the Session or SessionFilter for a Nvim instance."""
87	6	return self._session
88
89	6	def ui_attach(self, width, height, rgb):
90		"""Register as a remote UI.
91
92		After this method is called, the client will receive redraw
93		notifications.
94		"""
95		return self._session.request('ui_attach', width, height, rgb)
96
97	6	def ui_detach(self):
98		"""Unregister as a remote UI."""
99		return self._session.request('ui_detach')
100
101	6	def ui_try_resize(self, width, height):
102		"""Notify nvim that the client window has resized.
103
104		If possible, nvim will send a redraw request to resize.
105		"""
106		return self._session.request('ui_try_resize', width, height)
107
108	6	def subscribe(self, event):
109		"""Subscribe to a Nvim event."""
110	6	return self._session.request('vim_subscribe', event)
111
112	6	def unsubscribe(self, event):
113		"""Unsubscribe to a Nvim event."""
114	6	return self._session.request('vim_unsubscribe', event)
115
116	6	def command(self, string, async=False):
117		"""Execute a single ex command."""
118	6	return self._session.request('vim_command', string, async=async)
119
120	6	def command_output(self, string):
121		"""Execute a single ex command and return the output."""
122		return self._session.request('vim_command_output', string)
123
124	6	def eval(self, string, async=False):
125		"""Evaluate a vimscript expression."""
126	6	return self._session.request('vim_eval', string, async=async)
127
128	6	def call(self, name, args, *kwargs):
129		"""Call a vimscript function."""
130	6	for k in kwargs:
131	6	if k != "async":
132		raise TypeError(
133		"call() got an unexpected keyword argument '{}'".format(k))
134	6	return self._session.request('vim_call_function', name, args, **kwargs)
135
136	6	def strwidth(self, string):
137		"""Return the number of display cells `string` occupies.
138
139		Tab is counted as one cell.
140		"""
141	6	return self._session.request('vim_strwidth', string)
142
143	6	def list_runtime_paths(self):
144		"""Return a list of paths contained in the 'runtimepath' option."""
145		return self._session.request('vim_list_runtime_paths')
146
147	6	def foreach_rtp(self, cb):
148		"""Invoke `cb` for each path in 'runtimepath'.
149
150		Call the given callable for each path in 'runtimepath' until either
151		callable returns something but None, the exception is raised or there
152		are no longer paths. If stopped in case callable returned non-None,
153		vim.foreach_rtp function returns the value returned by callable.
154		"""
155		for path in self._session.request('vim_list_runtime_paths'):
156		try:
157		if cb(path) is not None:
158		break
159		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...
160		break
161
162	6	def chdir(self, dir_path):
163		"""Run os.chdir, then all appropriate vim stuff."""
164	6	os_chdir(dir_path)
165	6	return self._session.request('vim_change_directory', dir_path)
166
167	6	def feedkeys(self, keys, options='', escape_csi=True):
168		"""Push `keys` to Nvim user input buffer.
169
170		Options can be a string with the following character flags:
171		- 'm': Remap keys. This is default.
172		- 'n': Do not remap keys.
173		- 't': Handle keys as if typed; otherwise they are handled as if coming
174		from a mapping. This matters for undo, opening folds, etc.
175		"""
176		return self._session.request('vim_feedkeys', keys, options, escape_csi)
177
178	6	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...
179		"""Push `bytes` to Nvim low level input buffer.
180
181		Unlike `feedkeys()`, this uses the lowest level input buffer and the
182		call is not deferred. It returns the number of bytes actually
183		written(which can be less than what was requested if the buffer is
184		full).
185		"""
186	6	return self._session.request('vim_input', bytes)
187
188	6	def replace_termcodes(self, string, from_part=False, do_lt=True,
189		special=True):
190		r"""Replace any terminal code strings by byte sequences.
191
192		The returned sequences are Nvim's internal representation of keys,
193		for example:
194
195		<esc> -> '\x1b'
196		<cr> -> '\r'
197		<c-l> -> '\x0c'
198		<up> -> '\x80ku'
199
200		The returned sequences can be used as input to `feedkeys`.
201		"""
202		return self._session.request('vim_replace_termcodes', string,
203		from_part, do_lt, special)
204
205	6	def out_write(self, msg):
206		"""Print `msg` as a normal message."""
207		return self._session.request('vim_out_write', msg)
208
209	6	def err_write(self, msg, async=False):
210		"""Print `msg` as an error message."""
211		return self._session.request('vim_err_write', msg, async=async)
212
213	6	def quit(self, quit_command='qa!'):
214		"""Send a quit command to Nvim.
215
216		By default, the quit command is 'qa!' which will make Nvim quit without
217		saving anything.
218		"""
219		try:
220		self.command(quit_command)
221		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...
222		# sending a quit command will raise an IOError because the
223		# connection is closed before a response is received. Safe to
224		# ignore it.
225		pass
226
227	6	def new_highlight_source(self):
228		"""Return new src_id for use with Buffer.add_highlight."""
229		return self.current.buffer.add_highlight("", 0, src_id=0)
230
231	6	def async_call(self, fn, args, *kwargs):
232		"""Schedule `fn` to be called by the event loop soon.
233
234		This function is thread-safe, and is the only way code not
235		on the main thread could interact with nvim api objects.
236
237		This function can also be called in a synchronous
238		event handler, just before it returns, to defer execution
239		that shouldn't block neovim.
240		"""
241		call_point = ''.join(format_stack(None, 5)[:-1])
242
243		def handler():
244		try:
245		fn(args, *kwargs)
246		except Exception as err:
247		msg = ("error caught while executing async callback:\n"
248		"{!r}\n{}\n \nthe call was requested at\n{}"
249		.format(err, format_exc(5), call_point))
250		self.err_write(msg, async=True)
251		raise
252		self._session.threadsafe_call(handler)
253
254
255	6	class Current(object):
256
257		"""Helper class for emulating vim.current from python-vim."""
258
259	6	def __init__(self, session):
260	6	self._session = session
261	6	self.range = None
262
263	6	@property
264		def line(self):
265	6	return self._session.request('vim_get_current_line')
266
267	6	@line.setter
268		def line(self, line):
269	6	return self._session.request('vim_set_current_line', line)
270
271	6	@property
272		def buffer(self):
273	6	return self._session.request('vim_get_current_buffer')
274
275	6	@buffer.setter
276		def buffer(self, buffer):
277	6	return self._session.request('vim_set_current_buffer', buffer)
278
279	6	@property
280		def window(self):
281	6	return self._session.request('vim_get_current_window')
282
283	6	@window.setter
284		def window(self, window):
285	6	return self._session.request('vim_set_current_window', window)
286
287	6	@property
288		def tabpage(self):
289	6	return self._session.request('vim_get_current_tabpage')
290
291	6	@tabpage.setter
292		def tabpage(self, tabpage):
293	6	return self._session.request('vim_set_current_tabpage', tabpage)
294
295
296	6	class Funcs(object):
297
298		"""Helper class for functional vimscript interface."""
299
300	6	def __init__(self, nvim):
301	6	self._nvim = nvim
302
303	6	def __getattr__(self, name):
304	6	return functools.partial(self._nvim.call, name)
305
306
307	6	class ExtHook(SessionHook):
308	6	def __init__(self, types):
309	6	self.types = types
310	6	super(ExtHook, self).__init__(from_nvim=self.from_ext,
311		to_nvim=self.to_ext)
312
313	6	def from_ext(self, obj, session, method, kind):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `method` seems to be unused. Loading history... Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `kind` seems to be unused. Loading history...
314	6	if type(obj) is ExtType:
315	6	cls = self.types[obj.code]
316	6	return cls(session, (obj.code, obj.data))
317	6	return obj
318
319	6	def to_ext(self, obj, session, method, kind):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `method` seems to be unused. Loading history... Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `kind` seems to be unused. Loading history... Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `session` seems to be unused. Loading history... Coding Style introduced 2015-12-02 06:47 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...
320	6	if isinstance(obj, Remote):
321	6	return ExtType(*obj.code_data)
322	6	return obj
323
324
325	6	class NvimError(Exception):
326		pass
327

neovim / python-client

Pull Request — master (#169)

neovim.api.Nvim.input() A

Complexity

Size

Duplication

Code Coverage

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files