Host.predicate() - Code Metrics - Inspection of "asyncio: use stderr fd constant" - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#219)

by Björn

created 2016-08-20 09:20 UTC

Host.predicate() A

↳ Parent: Host._discover_functions()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

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

"""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
from ..util import format_exc_skip

__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 _wrap_function(self, fn, sync, decode, nvim_bind, name, *args):
        if decode:
            args = walk(decode_if_bytes, args, decode)
        if nvim_bind is not None:
            args.insert(0, nvim_bind)
        try:
            return fn(*args)
        except Exception:

            if sync:
                msg = ("error caught in request handler '{} {}':\n{}"
                       .format(name, args, format_exc_skip(1, 5)))
                raise ErrorResponse(msg)
            else:
                msg = ("error caught in async handler '{} {}'\n{}\n"
                       .format(name, args, format_exc_skip(1, 5)))
                self._on_async_err(msg + "\n")

    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)
        handler(*args)

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

        specs = []
        objdecode = getattr(obj, '_nvim_decode', self._decode_default)
        for _, fn in inspect.getmembers(obj, predicate):
            sync = 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
            decode = getattr(fn, '_nvim_decode', objdecode)
            nvim_bind = None
            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
                nvim_bind = self._configure_nvim_for(fn)

            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)

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

neovim / python-client

Pull Request — master (#219)

Host.predicate() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like