Host._on_notification() - Code Metrics - Inspection of "change default unicode error mode to "surrogateesc..." - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#199)

by Björn

created 2016-04-21 19:26 UTC

Host._on_notification() A

↳ Parent: Host

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	17.1835

Metric	Value
cc	4
dl	0
loc	19
ccs	1
cts	16
cp	0.0625
crap	17.1835
rs	9.2

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

from traceback import format_exc

from . import script_host
from ..api import decode_if_bytes, walk
from ..compat import IS_PYTHON3, find_module
from ..msgpack_rpc import ErrorResponse

__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._load_errors = {}
        self._notification_handlers = {}
        self._request_handlers = {
            'poll': lambda: 'ok',
            'specs': self._on_specs_request,
            'shutdown': self.shutdown
        }

        # Decode per default for Python3
        self._decode_default = IS_PYTHON3

    def _on_async_err(self, msg):
        self.nvim.err_write(msg, async=True)

    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),
                           err_cb=self._on_async_err)

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

    def _on_request(self, name, args):
        """Handle a msgpack-rpc request."""
        if IS_PYTHON3:
            name = decode_if_bytes(name)
        handler = self._request_handlers.get(name, None)
        if not handler:
            msg = self._missing_handler_error(name, 'request')
            error(msg)
            raise ErrorResponse(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:
            name = decode_if_bytes(name)
        handler = self._notification_handlers.get(name, None)
        if not handler:
            msg = self._missing_handler_error(name, 'notification')
            error(msg)
            self._on_async_err(msg + "\n")
            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{}\n"
                   .format(name, args, err, format_exc(5)))
            self._on_async_err(msg + "\n")
            raise

    def _missing_handler_error(self, name, kind):
        msg = 'no {} handler registered for "{}"'.format(kind, name)
        pathmatch = re.match(r'(.+):[^:]+:[^:]+', name)
        if pathmatch:
            loader_error = self._load_errors.get(pathmatch.group(1))
            if loader_error is not None:
                msg = msg + "\n" + loader_error
        return msg

    def _load(self, plugins):
        for path in plugins:
            err = None
            if path in self._loaded:
                error('{0} is already loaded'.format(path))
                continue
            try:
                if path == "script_host.py":
                    module = script_host
                else:
                    directory, name = os.path.split(os.path.splitext(path)[0])
                    file, pathname, descr = find_module(name, [directory])
                    module = imp.load_module(name, file, pathname, descr)
                handlers = []
                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 Exception as e:

                err = ('Encountered {} loading plugin at {}: {}\n{}'
                       .format(type(e).__name__, path, e, format_exc(5)))
                error(err)
                self._load_errors[path] = err

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

        def decoder(fn, decode, *args):
            return fn(*walk(decode_if_bytes, args, decode))
        specs = []
        objdecode = getattr(obj, '_nvim_decode', self._decode_default)
        for _, fn in inspect.getmembers(obj, predicate):
            decode = getattr(fn, '_nvim_decode', objdecode)
            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
            if decode:
                fn2 = functools.partial(decoder, fn, decode)
                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:
            path = decode_if_bytes(path)
        if path in self._load_errors:
            self.nvim.out_write(self._load_errors[path] + '\n')
        return self._specs.get(path, 0)

    def _configure_nvim_for(self, obj):
        # Configure a nvim instance for obj (checks encoding configuration)
        nvim = self.nvim
        decode = getattr(obj, '_nvim_decode', self._decode_default)
        if decode:
            nvim = nvim.with_decode(decode)
        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	6	import re
9
10	6	from traceback import format_exc
11
12	6	from . import script_host
13	6	from ..api import decode_if_bytes, walk
14	6	from ..compat import IS_PYTHON3, find_module
15	6	from ..msgpack_rpc import ErrorResponse
16
17	6	__all__ = ('Host')
18
19	6	logger = logging.getLogger(__name__)
20	6	error, debug, info, warn = (logger.error, logger.debug, logger.info,
21		logger.warning,)
22
23
24	6	class Host(object):
25
26		"""Nvim host for python plugins.
27
28		Takes care of loading/unloading plugins and routing msgpack-rpc
29		requests/notifications to the appropriate handlers.
30		"""
31
32	6	def __init__(self, nvim):
33		"""Set handlers for plugin_load/plugin_unload."""
34		self.nvim = nvim
35		self._specs = {}
36		self._loaded = {}
37		self._load_errors = {}
38		self._notification_handlers = {}
39		self._request_handlers = {
40		'poll': lambda: 'ok',
41		'specs': self._on_specs_request,
42		'shutdown': self.shutdown
43		}
44
45		# Decode per default for Python3
46		self._decode_default = IS_PYTHON3
47
48	6	def _on_async_err(self, msg):
49		self.nvim.err_write(msg, async=True)
50
51	6	def start(self, plugins):
52		"""Start listening for msgpack-rpc requests and notifications."""
53		self.nvim.run_loop(self._on_request,
54		self._on_notification,
55		lambda: self._load(plugins),
56		err_cb=self._on_async_err)
57
58	6	def shutdown(self):
59		"""Shutdown the host."""
60		self._unload()
61		self.nvim.stop_loop()
62
63	6	def _on_request(self, name, args):
64		"""Handle a msgpack-rpc request."""
65		if IS_PYTHON3:
66		name = decode_if_bytes(name)
67		handler = self._request_handlers.get(name, None)
68		if not handler:
69		msg = self._missing_handler_error(name, 'request')
70		error(msg)
71		raise ErrorResponse(msg)
72
73		debug('calling request handler for "%s", args: "%s"', name, args)
74		rv = handler(*args)
75		debug("request handler for '%s %s' returns: %s", name, args, rv)
76		return rv
77
78	6	def _on_notification(self, name, args):
79		"""Handle a msgpack-rpc notification."""
80		if IS_PYTHON3:
81		name = decode_if_bytes(name)
82		handler = self._notification_handlers.get(name, None)
83		if not handler:
84		msg = self._missing_handler_error(name, 'notification')
85		error(msg)
86		self._on_async_err(msg + "\n")
87		return
88
89		debug('calling notification handler for "%s", args: "%s"', name, args)
90		try:
91		handler(*args)
92		except Exception as err:
93		msg = ("error caught in async handler '{} {}':\n{!r}\n{}\n"
94		.format(name, args, err, format_exc(5)))
95		self._on_async_err(msg + "\n")
96		raise
97
98	6	def _missing_handler_error(self, name, kind):
99		msg = 'no {} handler registered for "{}"'.format(kind, name)
100		pathmatch = re.match(r'(.+):[^:]+:[^:]+', name)
101		if pathmatch:
102		loader_error = self._load_errors.get(pathmatch.group(1))
103		if loader_error is not None:
104		msg = msg + "\n" + loader_error
105		return msg
106
107	6	def _load(self, plugins):
108		for path in plugins:
109		err = None
110		if path in self._loaded:
111		error('{0} is already loaded'.format(path))
112		continue
113		try:
114		if path == "script_host.py":
115		module = script_host
116		else:
117		directory, name = os.path.split(os.path.splitext(path)[0])
118		file, pathname, descr = find_module(name, [directory])
119		module = imp.load_module(name, file, pathname, descr)
120		handlers = []
121		self._discover_classes(module, handlers, path)
122		self._discover_functions(module, handlers, path)
123		if not handlers:
124		error('{0} exports no handlers'.format(path))
125		continue
126		self._loaded[path] = {'handlers': handlers, 'module': module}
127		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...
128		err = ('Encountered {} loading plugin at {}: {}\n{}'
129		.format(type(e).__name__, path, e, format_exc(5)))
130		error(err)
131		self._load_errors[path] = err
132
133	6	def _unload(self):
134		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...
135		handlers = plugin['handlers']
136		for handler in handlers:
137		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...
138		if hasattr(handler, '_nvim_shutdown_hook'):
139		handler()
140		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...
141		del self._request_handlers[method_name]
142		else:
143		del self._notification_handlers[method_name]
144		self._specs = {}
145		self._loaded = {}
146
147	6	def _discover_classes(self, module, handlers, plugin_path):
148		for _, cls in inspect.getmembers(module, inspect.isclass):
149		if getattr(cls, '_nvim_plugin', False):
150		# create an instance of the plugin and pass the nvim object
151		plugin = cls(self._configure_nvim_for(cls))
152		# discover handlers in the plugin instance
153		self._discover_functions(plugin, handlers, plugin_path)
154
155	6	def _discover_functions(self, obj, handlers, plugin_path):
156		def predicate(o):
157		return hasattr(o, '_nvim_rpc_method_name')
158
159		def decoder(fn, decode, *args):
160		return fn(*walk(decode_if_bytes, args, decode))
161		specs = []
162		objdecode = getattr(obj, '_nvim_decode', self._decode_default)
163		for _, fn in inspect.getmembers(obj, predicate):
164		decode = getattr(fn, '_nvim_decode', objdecode)
165		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...
166		# bind a nvim instance to the handler
167		fn2 = functools.partial(fn, self._configure_nvim_for(fn))
168		# copy _nvim_* attributes from the original function
169		self._copy_attributes(fn, fn2)
170		fn = fn2
171		if decode:
172		fn2 = functools.partial(decoder, fn, decode)
173		self._copy_attributes(fn, fn2)
174		fn = fn2
175
176		# register in the rpc handler dict
177		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...
178		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...
179		method = '{0}:{1}'.format(plugin_path, method)
180		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...
181		if method in self._request_handlers:
182		raise Exception(('Request handler for "{0}" is ' +
183		'already registered').format(method))
184		self._request_handlers[method] = fn
185		else:
186		if method in self._notification_handlers:
187		raise Exception(('Notification handler for "{0}" is ' +
188		'already registered').format(method))
189		self._notification_handlers[method] = fn
190		if hasattr(fn, '_nvim_rpc_spec'):
191		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...
192		handlers.append(fn)
193		if specs:
194		self._specs[plugin_path] = specs
195
196	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...
197		# Copy _nvim_* attributes from the original function
198		for attr in dir(fn):
199		if attr.startswith('_nvim_'):
200		setattr(fn2, attr, getattr(fn, attr))
201
202	6	def _on_specs_request(self, path):
203		if IS_PYTHON3:
204		path = decode_if_bytes(path)
205		if path in self._load_errors:
206		self.nvim.out_write(self._load_errors[path] + '\n')
207		return self._specs.get(path, 0)
208
209	6	def _configure_nvim_for(self, obj):
210		# Configure a nvim instance for obj (checks encoding configuration)
211		nvim = self.nvim
212		decode = getattr(obj, '_nvim_decode', self._decode_default)
213		if decode:
214		nvim = nvim.with_decode(decode)
215		return nvim
216

neovim / python-client

Pull Request — master (#199)

Host._on_notification() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like