neovim.plugin.Host - Code Metrics - Inspection of "[WIP] better handling of exceptions in async conte..." - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#169)

by Björn

created 2016-02-24 17:44 UTC

neovim.plugin.Host C

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	179
Duplicated Lines	0 %

Test Coverage

Coverage

10.61%

Metric	Value
dl	0
loc	179
ccs	14
cts	132
cp	0.1061
rs	6
wmc	55

14 Methods

Rating	Name	Size	Complexity
A	predicate()	2	1
A	_copy_attributes()	5	3
A	start()	5	2
B	_unload()	13	5
F	_discover_functions()	39	12
B	_on_notification()	16	5
B	_decodehook_for()	7	5
A	_discover_classes()	7	3
A	_on_specs_request()	4	3
A	__init__()	14	4
A	_configure_nvim_for()	8	2
A	_on_request()	14	4
B	_load()	22	6
A	shutdown()	4	1

How to fix Complexity

"""Implements a Nvim host for python plugins."""
import functools
import imp
import inspect
import logging
import os
import os.path

from traceback import format_exc

from ..api import DecodeHook
from ..compat import IS_PYTHON3, find_module

__all__ = ('Host')

logger = logging.getLogger(__name__)
error, debug, info, warn = (logger.error, logger.debug, logger.info,
                            logger.warning,)


class Host(object):

    """Nvim host for python plugins.

    Takes care of loading/unloading plugins and routing msgpack-rpc
    requests/notifications to the appropriate handlers.
    """

    def __init__(self, nvim):
        """Set handlers for plugin_load/plugin_unload."""
        self.nvim = nvim
        self._specs = {}
        self._loaded = {}
        self._notification_handlers = {}
        self._request_handlers = {
            'poll': lambda: 'ok',
            'specs': self._on_specs_request,
            'shutdown': self.shutdown
        }
        self._nvim_encoding = nvim.options['encoding']
        if IS_PYTHON3 and isinstance(self._nvim_encoding, bytes):
            self._nvim_encoding = self._nvim_encoding.decode('ascii')

    def start(self, plugins):
        """Start listening for msgpack-rpc requests and notifications."""
        self.nvim.session.run(self._on_request,
                              self._on_notification,
                              lambda: self._load(plugins))

    def shutdown(self):
        """Shutdown the host."""
        self._unload()
        self.nvim.session.stop()

    def _on_request(self, name, args):
        """Handle a msgpack-rpc request."""
        if IS_PYTHON3 and isinstance(name, bytes):
            name = name.decode(self._nvim_encoding)
        handler = self._request_handlers.get(name, None)
        if not handler:
            msg = 'no request handler registered for "%s"' % name
            warn(msg)
            raise Exception(msg)

        debug('calling request handler for "%s", args: "%s"', name, args)
        rv = handler(*args)
        debug("request handler for '%s %s' returns: %s", name, args, rv)
        return rv

    def _on_notification(self, name, args):
        """Handle a msgpack-rpc notification."""
        if IS_PYTHON3 and isinstance(name, bytes):
            name = name.decode(self._nvim_encoding)
        handler = self._notification_handlers.get(name, None)
        if not handler:
            warn('no notification handler registered for "%s"', name)
            return

        debug('calling notification handler for "%s", args: "%s"', name, args)
        try:
            handler(*args)
        except Exception as err:
            msg = "error caught in async handler '{} {}':\n{!r}\n{}".format(name, args, err, format_exc(5))
            self.nvim.err_write(msg,async=True)
            raise


    def _load(self, plugins):
        for path in plugins:
            if path in self._loaded:
                error('{0} is already loaded'.format(path))
                continue
            directory, name = os.path.split(os.path.splitext(path)[0])
            file, pathname, description = find_module(name, [directory])
            handlers = []
            try:
                module = imp.load_module(name, file, pathname, description)
                self._discover_classes(module, handlers, path)
                self._discover_functions(module, handlers, path)
                if not handlers:
                    error('{0} exports no handlers'.format(path))
                    continue
                self._loaded[path] = {'handlers': handlers, 'module': module}
            except (ImportError, SyntaxError) as e:
                error(('Encountered import error loading '
                       'plugin at {0}: {1}').format(path, e))
            except Exception as e:

                error('Error loading plugin at {0} {1}: {2}'.format(
                    path, type(e).__name__, e))

    def _unload(self):
        for path, plugin in self._loaded.items():

            handlers = plugin['handlers']
            for handler in handlers:
                method_name = handler._nvim_rpc_method_name
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
                if hasattr(handler, '_nvim_shutdown_hook'):
                    handler()
                elif handler._nvim_rpc_sync:
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
                    del self._request_handlers[method_name]
                else:
                    del self._notification_handlers[method_name]
        self._specs = {}
        self._loaded = {}

    def _discover_classes(self, module, handlers, plugin_path):
        for _, cls in inspect.getmembers(module, inspect.isclass):
            if getattr(cls, '_nvim_plugin', False):
                # create an instance of the plugin and pass the nvim object
                plugin = cls(self._configure_nvim_for(cls))
                # discover handlers in the plugin instance
                self._discover_functions(plugin, handlers, plugin_path)

    def _discover_functions(self, obj, handlers, plugin_path):
        def predicate(o):
            return hasattr(o, '_nvim_rpc_method_name')
        specs = []
        objenc = getattr(obj, '_nvim_encoding', None)
        for _, fn in inspect.getmembers(obj, predicate):
            enc = getattr(fn, '_nvim_encoding', objenc)
            if fn._nvim_bind:
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
                # bind a nvim instance to the handler
                fn2 = functools.partial(fn, self._configure_nvim_for(fn))
                # copy _nvim_* attributes from the original function
                self._copy_attributes(fn, fn2)
                fn = fn2
            decodehook = self._decodehook_for(enc)
            if decodehook is not None:
                decoder = lambda fn, hook, *args: fn(*hook.walk(args))
                fn2 = functools.partial(decoder, fn, decodehook)
                self._copy_attributes(fn, fn2)
                fn = fn2

            # register in the rpc handler dict
            method = fn._nvim_rpc_method_name
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
            if fn._nvim_prefix_plugin_path:
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
                method = '{0}:{1}'.format(plugin_path, method)
            if fn._nvim_rpc_sync:
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
                if method in self._request_handlers:
                    raise Exception(('Request handler for "{0}" is ' +
                                    'already registered').format(method))
                self._request_handlers[method] = fn
            else:
                if method in self._notification_handlers:
                    raise Exception(('Notification handler for "{0}" is ' +
                                    'already registered').format(method))
                self._notification_handlers[method] = fn
            if hasattr(fn, '_nvim_rpc_spec'):
                specs.append(fn._nvim_rpc_spec)
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
            handlers.append(fn)
        if specs:
            self._specs[plugin_path] = specs

    def _copy_attributes(self, fn, fn2):
class Foo:
    def some_method(self, x, y):
        return x + y;
        # Copy _nvim_* attributes from the original function
        for attr in dir(fn):
            if attr.startswith('_nvim_'):
                setattr(fn2, attr, getattr(fn, attr))

    def _on_specs_request(self, path):
        if IS_PYTHON3 and isinstance(path, bytes):
            path = path.decode(self._nvim_encoding)
        return self._specs.get(path, [])

    def _decodehook_for(self, encoding):
        if IS_PYTHON3 and encoding is None:
            encoding = True
        if encoding is True:
            encoding = self._nvim_encoding
        if encoding:
            return DecodeHook(encoding)

    def _configure_nvim_for(self, obj):
        # Configure a nvim instance for obj (checks encoding configuration)
        nvim = self.nvim
        encoding = getattr(obj, '_nvim_encoding', None)
        hook = self._decodehook_for(encoding)
        if hook is not None:
            nvim = nvim.with_hook(hook)
        return nvim


1		"""Implements a Nvim host for python plugins."""
2	6	import functools
3	6	import imp
4	6	import inspect
5	6	import logging
6	6	import os
7	6	import os.path
8
9	6	from traceback import format_exc
10
11	6	from ..api import DecodeHook
12	6	from ..compat import IS_PYTHON3, find_module
13
14	6	__all__ = ('Host')
15
16	6	logger = logging.getLogger(__name__)
17	6	error, debug, info, warn = (logger.error, logger.debug, logger.info,
18		logger.warning,)
19
20
21	6	class Host(object):
22
23		"""Nvim host for python plugins.
24
25		Takes care of loading/unloading plugins and routing msgpack-rpc
26		requests/notifications to the appropriate handlers.
27		"""
28
29	6	def __init__(self, nvim):
30		"""Set handlers for plugin_load/plugin_unload."""
31		self.nvim = nvim
32		self._specs = {}
33		self._loaded = {}
34		self._notification_handlers = {}
35		self._request_handlers = {
36		'poll': lambda: 'ok',
37		'specs': self._on_specs_request,
38		'shutdown': self.shutdown
39		}
40		self._nvim_encoding = nvim.options['encoding']
41		if IS_PYTHON3 and isinstance(self._nvim_encoding, bytes):
42		self._nvim_encoding = self._nvim_encoding.decode('ascii')
43
44	6	def start(self, plugins):
45		"""Start listening for msgpack-rpc requests and notifications."""
46		self.nvim.session.run(self._on_request,
47		self._on_notification,
48		lambda: self._load(plugins))
49
50	6	def shutdown(self):
51		"""Shutdown the host."""
52		self._unload()
53		self.nvim.session.stop()
54
55	6	def _on_request(self, name, args):
56		"""Handle a msgpack-rpc request."""
57		if IS_PYTHON3 and isinstance(name, bytes):
58		name = name.decode(self._nvim_encoding)
59		handler = self._request_handlers.get(name, None)
60		if not handler:
61		msg = 'no request handler registered for "%s"' % name
62		warn(msg)
63		raise Exception(msg)
64
65		debug('calling request handler for "%s", args: "%s"', name, args)
66		rv = handler(*args)
67		debug("request handler for '%s %s' returns: %s", name, args, rv)
68		return rv
69
70	6	def _on_notification(self, name, args):
71		"""Handle a msgpack-rpc notification."""
72		if IS_PYTHON3 and isinstance(name, bytes):
73		name = name.decode(self._nvim_encoding)
74		handler = self._notification_handlers.get(name, None)
75		if not handler:
76		warn('no notification handler registered for "%s"', name)
77		return
78
79		debug('calling notification handler for "%s", args: "%s"', name, args)
80		try:
81		handler(*args)
82		except Exception as err:
83		msg = "error caught in async handler '{} {}':\n{!r}\n{}".format(name, args, err, format_exc(5))
84		self.nvim.err_write(msg,async=True)
85		raise
86
87
88	6	def _load(self, plugins):
89		for path in plugins:
90		if path in self._loaded:
91		error('{0} is already loaded'.format(path))
92		continue
93		directory, name = os.path.split(os.path.splitext(path)[0])
94		file, pathname, description = find_module(name, [directory])
95		handlers = []
96		try:
97		module = imp.load_module(name, file, pathname, description)
98		self._discover_classes(module, handlers, path)
99		self._discover_functions(module, handlers, path)
100		if not handlers:
101		error('{0} exports no handlers'.format(path))
102		continue
103		self._loaded[path] = {'handlers': handlers, 'module': module}
104		except (ImportError, SyntaxError) as e:
105		error(('Encountered import error loading '
106		'plugin at {0}: {1}').format(path, e))
107		except Exception as e:
		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...
108		error('Error loading plugin at {0} {1}: {2}'.format(
109		path, type(e).__name__, e))
110
111	6	def _unload(self):
112		for path, plugin in self._loaded.items():
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The variable `path` seems to be unused. Loading history...
113		handlers = plugin['handlers']
114		for handler in handlers:
115		method_name = handler._nvim_rpc_method_name
		0 ignored issues – show Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_rpc_method_name` 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...
116		if hasattr(handler, '_nvim_shutdown_hook'):
117		handler()
118		elif handler._nvim_rpc_sync:
		0 ignored issues – show Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_rpc_sync` 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...
119		del self._request_handlers[method_name]
120		else:
121		del self._notification_handlers[method_name]
122		self._specs = {}
123		self._loaded = {}
124
125	6	def _discover_classes(self, module, handlers, plugin_path):
126		for _, cls in inspect.getmembers(module, inspect.isclass):
127		if getattr(cls, '_nvim_plugin', False):
128		# create an instance of the plugin and pass the nvim object
129		plugin = cls(self._configure_nvim_for(cls))
130		# discover handlers in the plugin instance
131		self._discover_functions(plugin, handlers, plugin_path)
132
133	6	def _discover_functions(self, obj, handlers, plugin_path):
134		def predicate(o):
135		return hasattr(o, '_nvim_rpc_method_name')
136		specs = []
137		objenc = getattr(obj, '_nvim_encoding', None)
138		for _, fn in inspect.getmembers(obj, predicate):
139		enc = getattr(fn, '_nvim_encoding', objenc)
140		if fn._nvim_bind:
		0 ignored issues – show Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_bind` 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...
141		# bind a nvim instance to the handler
142		fn2 = functools.partial(fn, self._configure_nvim_for(fn))
143		# copy _nvim_* attributes from the original function
144		self._copy_attributes(fn, fn2)
145		fn = fn2
146		decodehook = self._decodehook_for(enc)
147		if decodehook is not None:
148		decoder = lambda fn, hook, args: fn(hook.walk(args))
149		fn2 = functools.partial(decoder, fn, decodehook)
150		self._copy_attributes(fn, fn2)
151		fn = fn2
152
153		# register in the rpc handler dict
154		method = fn._nvim_rpc_method_name
		0 ignored issues – show Bug introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The Function `newfunc` does not seem to have a member named `_nvim_rpc_method_name`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history... Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_rpc_method_name` 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...
155		if fn._nvim_prefix_plugin_path:
		0 ignored issues – show Bug introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The Function `newfunc` does not seem to have a member named `_nvim_prefix_plugin_path`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history... Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_prefix_plugin_path` 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...
156		method = '{0}:{1}'.format(plugin_path, method)
157		if fn._nvim_rpc_sync:
		0 ignored issues – show Bug introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The Function `newfunc` does not seem to have a member named `_nvim_rpc_sync`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history... Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_rpc_sync` 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...
158		if method in self._request_handlers:
159		raise Exception(('Request handler for "{0}" is ' +
160		'already registered').format(method))
161		self._request_handlers[method] = fn
162		else:
163		if method in self._notification_handlers:
164		raise Exception(('Notification handler for "{0}" is ' +
165		'already registered').format(method))
166		self._notification_handlers[method] = fn
167		if hasattr(fn, '_nvim_rpc_spec'):
168		specs.append(fn._nvim_rpc_spec)
		0 ignored issues – show Bug introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The Function `newfunc` does not seem to have a member named `_nvim_rpc_spec`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history... Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_rpc_spec` 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...
169		handlers.append(fn)
170		if specs:
171		self._specs[plugin_path] = specs
172
173	6	def _copy_attributes(self, fn, fn2):
		0 ignored issues – show 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...
174		# Copy _nvim_* attributes from the original function
175		for attr in dir(fn):
176		if attr.startswith('_nvim_'):
177		setattr(fn2, attr, getattr(fn, attr))
178
179	6	def _on_specs_request(self, path):
180		if IS_PYTHON3 and isinstance(path, bytes):
181		path = path.decode(self._nvim_encoding)
182		return self._specs.get(path, [])
183
184	6	def _decodehook_for(self, encoding):
185		if IS_PYTHON3 and encoding is None:
186		encoding = True
187		if encoding is True:
188		encoding = self._nvim_encoding
189		if encoding:
190		return DecodeHook(encoding)
191
192	6	def _configure_nvim_for(self, obj):
193		# Configure a nvim instance for obj (checks encoding configuration)
194		nvim = self.nvim
195		encoding = getattr(obj, '_nvim_encoding', None)
196		hook = self._decodehook_for(encoding)
197		if hook is not None:
198		nvim = nvim.with_hook(hook)
199		return nvim
200

neovim / python-client

Pull Request — master (#169)

neovim.plugin.Host C

Complexity

Size/Duplication

Test Coverage

14 Methods

How to fix Complexity

Complex Class

Duplication Side-by-Side

Filter issues like