neovim.plugin.Host._load() - Code Metrics - Inspection of "Refactor: let Nvim be the session wrapper and move..." - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#179)

by Björn

created 2016-03-08 14:19 UTC

neovim.plugin.Host._load() C

↳ Parent: neovim.plugin.Host

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	44.2913

Metric	Value
cc	7
dl	0
loc	26
ccs	2
cts	23
cp	0.0869
crap	44.2913
rs	5.5

"""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.run_loop(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
            if path == "script_host.py":
                directory = os.path.dirname(__file__)
                name = "script_host"
            else:
                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_decodehook(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.run_loop(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{}"
84		.format(name, args, err, format_exc(5)))
85		self.nvim.err_write(msg, async=True)
86		raise
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		if path == "script_host.py":
94		directory = os.path.dirname(__file__)
95		name = "script_host"
96		else:
97		directory, name = os.path.split(os.path.splitext(path)[0])
98		file, pathname, description = find_module(name, [directory])
99		handlers = []
100		try:
101		module = imp.load_module(name, file, pathname, description)
102		self._discover_classes(module, handlers, path)
103		self._discover_functions(module, handlers, path)
104		if not handlers:
105		error('{0} exports no handlers'.format(path))
106		continue
107		self._loaded[path] = {'handlers': handlers, 'module': module}
108		except (ImportError, SyntaxError) as e:
109		error(('Encountered import error loading '
110		'plugin at {0}: {1}').format(path, e))
111	6	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...
112		error('Error loading plugin at {0} {1}: {2}'.format(
113		path, type(e).__name__, e))
114
115		def _unload(self):
116		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...
117		handlers = plugin['handlers']
118		for handler in handlers:
119		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...
120		if hasattr(handler, '_nvim_shutdown_hook'):
121		handler()
122		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...
123		del self._request_handlers[method_name]
124		else:
125	6	del self._notification_handlers[method_name]
126		self._specs = {}
127		self._loaded = {}
128
129		def _discover_classes(self, module, handlers, plugin_path):
130		for _, cls in inspect.getmembers(module, inspect.isclass):
131		if getattr(cls, '_nvim_plugin', False):
132		# create an instance of the plugin and pass the nvim object
133	6	plugin = cls(self._configure_nvim_for(cls))
134		# discover handlers in the plugin instance
135		self._discover_functions(plugin, handlers, plugin_path)
136
137		def _discover_functions(self, obj, handlers, plugin_path):
138		def predicate(o):
139		return hasattr(o, '_nvim_rpc_method_name')
140		specs = []
141		objenc = getattr(obj, '_nvim_encoding', None)
142		for _, fn in inspect.getmembers(obj, predicate):
143		enc = getattr(fn, '_nvim_encoding', objenc)
144		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...
145		# bind a nvim instance to the handler
146		fn2 = functools.partial(fn, self._configure_nvim_for(fn))
147		# copy _nvim_* attributes from the original function
148		self._copy_attributes(fn, fn2)
149		fn = fn2
150		decodehook = self._decodehook_for(enc)
151		if decodehook is not None:
152		decoder = lambda fn, hook, args: fn(hook.walk(args))
153		fn2 = functools.partial(decoder, fn, decodehook)
154		self._copy_attributes(fn, fn2)
155		fn = fn2
156
157		# register in the rpc handler dict
158		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...
159		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...
160		method = '{0}:{1}'.format(plugin_path, method)
161		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...
162		if method in self._request_handlers:
163		raise Exception(('Request handler for "{0}" is ' +
164		'already registered').format(method))
165		self._request_handlers[method] = fn
166		else:
167		if method in self._notification_handlers:
168		raise Exception(('Notification handler for "{0}" is ' +
169		'already registered').format(method))
170		self._notification_handlers[method] = fn
171		if hasattr(fn, '_nvim_rpc_spec'):
172		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...
173	6	handlers.append(fn)
174		if specs:
175		self._specs[plugin_path] = specs
176
177		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...
178		# Copy _nvim_* attributes from the original function
179	6	for attr in dir(fn):
180		if attr.startswith('_nvim_'):
181		setattr(fn2, attr, getattr(fn, attr))
182
183		def _on_specs_request(self, path):
184	6	if IS_PYTHON3 and isinstance(path, bytes):
185		path = path.decode(self._nvim_encoding)
186		return self._specs.get(path, [])
187
188		def _decodehook_for(self, encoding):
189		if IS_PYTHON3 and encoding is None:
190		encoding = True
191		if encoding is True:
192	6	encoding = self._nvim_encoding
193		if encoding:
194		return DecodeHook(encoding)
195
196		def _configure_nvim_for(self, obj):
197		# Configure a nvim instance for obj (checks encoding configuration)
198		nvim = self.nvim
199		encoding = getattr(obj, '_nvim_encoding', None)
200		hook = self._decodehook_for(encoding)
201		if hook is not None:
202		nvim = nvim.with_decodehook(hook)
203		return nvim
204

neovim / python-client

Pull Request — master (#179)

neovim.plugin.Host._load() C

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like