neovim /
python-client
| Total Complexity | 58 |
| Total Lines | 194 |
| Duplicated Lines | 0 % |
| Coverage | 9.92% |
Complex classes like neovim.plugin.Host often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
| 1 | """Implements a Nvim host for python plugins.""" |
||
| 23 | class Host(object): |
||
| 24 | |||
| 25 | """Nvim host for python plugins. |
||
| 26 | |||
| 27 | Takes care of loading/unloading plugins and routing msgpack-rpc |
||
| 28 | requests/notifications to the appropriate handlers. |
||
| 29 | 6 | """ |
|
| 30 | |||
| 31 | def __init__(self, nvim): |
||
| 32 | """Set handlers for plugin_load/plugin_unload.""" |
||
| 33 | self.nvim = nvim |
||
| 34 | self._specs = {} |
||
| 35 | self._loaded = {} |
||
| 36 | self._load_errors = {} |
||
| 37 | self._notification_handlers = {} |
||
| 38 | self._request_handlers = { |
||
| 39 | 'poll': lambda: 'ok', |
||
| 40 | 'specs': self._on_specs_request, |
||
| 41 | 'shutdown': self.shutdown |
||
| 42 | } |
||
| 43 | self._nvim_encoding = nvim.options['encoding'] |
||
| 44 | 6 | if IS_PYTHON3 and isinstance(self._nvim_encoding, bytes): |
|
| 45 | self._nvim_encoding = self._nvim_encoding.decode('ascii') |
||
| 46 | |||
| 47 | def start(self, plugins): |
||
| 48 | """Start listening for msgpack-rpc requests and notifications.""" |
||
| 49 | self.nvim.session.run(self._on_request, |
||
| 50 | 6 | self._on_notification, |
|
| 51 | lambda: self._load(plugins)) |
||
| 52 | |||
| 53 | def shutdown(self): |
||
| 54 | """Shutdown the host.""" |
||
| 55 | 6 | self._unload() |
|
| 56 | self.nvim.session.stop() |
||
| 57 | |||
| 58 | def _on_request(self, name, args): |
||
| 59 | """Handle a msgpack-rpc request.""" |
||
| 60 | if IS_PYTHON3 and isinstance(name, bytes): |
||
| 61 | name = name.decode(self._nvim_encoding) |
||
| 62 | handler = self._request_handlers.get(name, None) |
||
| 63 | if not handler: |
||
| 64 | msg = self.missing_handler_error(name, "notification") |
||
| 65 | error(msg) |
||
| 66 | raise ErrorResponse(msg) |
||
| 67 | |||
| 68 | debug('calling request handler for "%s", args: "%s"', name, args) |
||
| 69 | rv = handler(*args) |
||
| 70 | 6 | debug("request handler for '%s %s' returns: %s", name, args, rv) |
|
| 71 | return rv |
||
| 72 | |||
| 73 | def _on_notification(self, name, args): |
||
| 74 | """Handle a msgpack-rpc notification.""" |
||
| 75 | if IS_PYTHON3 and isinstance(name, bytes): |
||
| 76 | name = name.decode(self._nvim_encoding) |
||
| 77 | handler = self._notification_handlers.get(name, None) |
||
| 78 | if not handler: |
||
| 79 | msg = self.missing_handler_error(name, "notification") |
||
| 80 | error(msg) |
||
| 81 | self.nvim.err_write(msg+"\n") |
||
| 82 | return |
||
| 83 | |||
| 84 | debug('calling notification handler for "%s", args: "%s"', name, args) |
||
| 85 | try: |
||
| 86 | handler(*args) |
||
| 87 | except Exception as err: |
||
| 88 | 6 | msg = ("error caught in async handler '{} {}':\n{!r}\n{}" |
|
| 89 | .format(name, args, err, format_exc(5))) |
||
| 90 | self.nvim.err_write(msg, async=True) |
||
| 91 | raise |
||
| 92 | |||
| 93 | def missing_handler_error(self, name, kind): |
||
| 94 | msg = 'no {} handler registered for "{}"'.format(kind, name) |
||
| 95 | last_colon = name.rfind(':') |
||
|
Unused Code
introduced
by
|
|||
| 96 | pathmatch = re.match(r'(.+):[^:]+:[^:]+', name) |
||
| 97 | if pathmatch: |
||
| 98 | loader_error = self._load_errors.get(pathmatch.group(1)) |
||
| 99 | if loader_error is not None: |
||
| 100 | msg = msg + "\n" + loader_error |
||
| 101 | return msg |
||
| 102 | |||
| 103 | def _load(self, plugins): |
||
| 104 | for path in plugins: |
||
| 105 | err = None |
||
| 106 | if path in self._loaded: |
||
| 107 | error('{0} is already loaded'.format(path)) |
||
| 108 | continue |
||
| 109 | directory, name = os.path.split(os.path.splitext(path)[0]) |
||
| 110 | file, pathname, description = find_module(name, [directory]) |
||
| 111 | 6 | handlers = [] |
|
| 112 | try: |
||
| 113 | module = imp.load_module(name, file, pathname, description) |
||
| 114 | self._discover_classes(module, handlers, path) |
||
| 115 | self._discover_functions(module, handlers, path) |
||
| 116 | if not handlers: |
||
| 117 | error('{0} exports no handlers'.format(path)) |
||
| 118 | continue |
||
| 119 | self._loaded[path] = {'handlers': handlers, 'module': module} |
||
| 120 | except Exception as e: |
||
| 121 | err = ('Encountered {} loading plugin at {}: {}\n{}' |
||
| 122 | .format(type(e).__name__, path, e, format_exc(5))) |
||
| 123 | error(err) |
||
| 124 | self._load_errors[path] = err |
||
| 125 | 6 | ||
| 126 | def _unload(self): |
||
| 127 | for path, plugin in self._loaded.items(): |
||
| 128 | handlers = plugin['handlers'] |
||
| 129 | for handler in handlers: |
||
| 130 | method_name = handler._nvim_rpc_method_name |
||
| 131 | if hasattr(handler, '_nvim_shutdown_hook'): |
||
| 132 | handler() |
||
| 133 | 6 | elif handler._nvim_rpc_sync: |
|
| 134 | del self._request_handlers[method_name] |
||
| 135 | else: |
||
| 136 | del self._notification_handlers[method_name] |
||
| 137 | self._specs = {} |
||
| 138 | self._loaded = {} |
||
| 139 | |||
| 140 | def _discover_classes(self, module, handlers, plugin_path): |
||
| 141 | for _, cls in inspect.getmembers(module, inspect.isclass): |
||
| 142 | if getattr(cls, '_nvim_plugin', False): |
||
| 143 | # create an instance of the plugin and pass the nvim object |
||
| 144 | plugin = cls(self._configure_nvim_for(cls)) |
||
| 145 | # discover handlers in the plugin instance |
||
| 146 | self._discover_functions(plugin, handlers, plugin_path) |
||
| 147 | |||
| 148 | def _discover_functions(self, obj, handlers, plugin_path): |
||
| 149 | def predicate(o): |
||
| 150 | return hasattr(o, '_nvim_rpc_method_name') |
||
| 151 | specs = [] |
||
| 152 | objenc = getattr(obj, '_nvim_encoding', None) |
||
| 153 | for _, fn in inspect.getmembers(obj, predicate): |
||
| 154 | enc = getattr(fn, '_nvim_encoding', objenc) |
||
| 155 | if fn._nvim_bind: |
||
| 156 | # bind a nvim instance to the handler |
||
| 157 | fn2 = functools.partial(fn, self._configure_nvim_for(fn)) |
||
| 158 | # copy _nvim_* attributes from the original function |
||
| 159 | self._copy_attributes(fn, fn2) |
||
| 160 | fn = fn2 |
||
| 161 | decodehook = self._decodehook_for(enc) |
||
| 162 | if decodehook is not None: |
||
| 163 | decoder = lambda fn, hook, *args: fn(*hook.walk(args)) |
||
| 164 | fn2 = functools.partial(decoder, fn, decodehook) |
||
| 165 | self._copy_attributes(fn, fn2) |
||
| 166 | fn = fn2 |
||
| 167 | |||
| 168 | # register in the rpc handler dict |
||
| 169 | method = fn._nvim_rpc_method_name |
||
| 170 | if fn._nvim_prefix_plugin_path: |
||
| 171 | method = '{0}:{1}'.format(plugin_path, method) |
||
| 172 | if fn._nvim_rpc_sync: |
||
| 173 | 6 | if method in self._request_handlers: |
|
| 174 | raise Exception(('Request handler for "{0}" is ' + |
||
| 175 | 'already registered').format(method)) |
||
| 176 | self._request_handlers[method] = fn |
||
| 177 | else: |
||
| 178 | if method in self._notification_handlers: |
||
| 179 | 6 | raise Exception(('Notification handler for "{0}" is ' + |
|
| 180 | 'already registered').format(method)) |
||
| 181 | self._notification_handlers[method] = fn |
||
| 182 | if hasattr(fn, '_nvim_rpc_spec'): |
||
| 183 | specs.append(fn._nvim_rpc_spec) |
||
| 184 | 6 | handlers.append(fn) |
|
| 185 | if specs: |
||
| 186 | self._specs[plugin_path] = specs |
||
| 187 | |||
| 188 | def _copy_attributes(self, fn, fn2): |
||
| 189 | # Copy _nvim_* attributes from the original function |
||
| 190 | for attr in dir(fn): |
||
| 191 | if attr.startswith('_nvim_'): |
||
| 192 | 6 | setattr(fn2, attr, getattr(fn, attr)) |
|
| 193 | |||
| 194 | def _on_specs_request(self, path): |
||
| 195 | if IS_PYTHON3 and isinstance(path, bytes): |
||
| 196 | path = path.decode(self._nvim_encoding) |
||
| 197 | if self._load_errors: |
||
| 198 | self.nvim.out_write('\n'.join(self._load_errors.values())+'\n') |
||
| 199 | return self._specs.get(path, []) |
||
| 200 | |||
| 201 | def _decodehook_for(self, encoding): |
||
| 202 | if IS_PYTHON3 and encoding is None: |
||
| 203 | encoding = True |
||
| 204 | if encoding is True: |
||
| 205 | encoding = self._nvim_encoding |
||
| 206 | if encoding: |
||
| 207 | return DecodeHook(encoding) |
||
| 208 | |||
| 209 | def _configure_nvim_for(self, obj): |
||
| 210 | # Configure a nvim instance for obj (checks encoding configuration) |
||
| 211 | nvim = self.nvim |
||
| 212 | encoding = getattr(obj, '_nvim_encoding', None) |
||
| 213 | hook = self._decodehook_for(encoding) |
||
| 214 | if hook is not None: |
||
| 215 | nvim = nvim.with_hook(hook) |
||
| 216 | return nvim |
||
| 217 |
Loading history...
