Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""Event loop abstraction subpackage.
Tries to use pyuv as a backend, falling back to the asyncio implementation.
"""
from ...compat import IS_PYTHON3
# on python3 we only support asyncio, as we expose it to plugins
if IS_PYTHON3:
from .asyncio import AsyncioEventLoop
EventLoop = AsyncioEventLoop
else:
try:
# libuv is fully implemented in C, use it when available
from .uv import UvEventLoop
EventLoop = UvEventLoop
except ImportError:
# asyncio(trollius on python 2) is pure python and should be more
# portable across python implementations
from .asyncio import AsyncioEventLoop
EventLoop = AsyncioEventLoop
__all__ = ('EventLoop')
"""Set handlers for plugin_load/plugin_unload."""
self.nvim = nvim
self._specs = {}
self._loaded = {}
self._load_errors = {}
self._notification_handlers = {
'nvim_error_event': self._on_error_event
}
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 from_session(cls, session):
"""Create a new Nvim instance for a Session instance.
This method must be called to create the first Nvim instance, since it
queries Nvim metadata for type information and sets a SessionHook for
creating specialized objects from Nvim remote handles.
"""
session.error_wrapper = lambda e: NvimError(decode_if_bytes(e[1]))
channel_id, metadata = session.request(b'nvim_get_api_info')
if IS_PYTHON3:
# decode all metadata strings for python3
metadata = walk(decode_if_bytes, metadata)
types = {
metadata['types']['Buffer']['id']: Buffer,
metadata['types']['Window']['id']: Window,
metadata['types']['Tabpage']['id']: Tabpage,
}
return cls(session, channel_id, metadata, types)
def plugin(cls):
"""Tag a class as a plugin.
This decorator is required to make the class methods discoverable by the
plugin_load method of the host.
"""
cls._nvim_plugin = True
# the _nvim_bind attribute is set to True by default, meaning that
# decorated functions have a bound Nvim instance as first argument.
# For methods in a plugin-decorated class this is not required, because
# the class initializer will already receive the nvim object.
predicate = lambda fn: hasattr(fn, '_nvim_bind')
for _, fn in inspect.getmembers(cls, predicate):
if IS_PYTHON3:
fn._nvim_bind = False
else:
fn.im_func._nvim_bind = False
return cls
This will automatically close the session when you're done with it, or
when an error occured.
"""
session = (tcp_session(address, port) if session_type == 'tcp' else
socket_session(path) if session_type == 'socket' else
stdio_session() if session_type == 'stdio' else
child_session(argv) if session_type == 'child' else
None)
if not session:
raise Exception('Unknown session type "%s"' % session_type)
if decode is None:
decode = IS_PYTHON3
return Nvim.from_session(session).with_decode(decode)
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
"""API for working with a Nvim Buffer."""
from .common import Remote
from ..compat import IS_PYTHON3, check_async
__all__ = ('Buffer')
if IS_PYTHON3:
basestring = str
def adjust_index(idx, default=None):
"""Convert from python indexing convention to nvim indexing convention."""
if idx is None:
return default
elif idx < 0:
return idx - 1
else:
return idx
class Buffer(Remote):
"""A remote Nvim buffer."""