ScriptHost.python_execute_async() - Code Metrics - Inspection of "[WIP] feature: add notification handler for python..." - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#273)

unknown

created 2017-10-13 02:37 UTC

ScriptHost.python_execute_async() A

↳ Parent: ScriptHost

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes	1
Bugs	0	Features	1

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

"""Legacy python/python3-vim emulation."""
import imp
import io
import logging
import os
import sys

from .decorators import plugin, rpc_export
from ..api import Nvim, walk
from ..msgpack_rpc import ErrorResponse
from ..util import format_exc_skip

__all__ = ('ScriptHost',)


logger = logging.getLogger(__name__)
debug, info, warn, exception = (logger.debug,
                                logger.info,
                                logger.warn,
                                logger.exception)

IS_PYTHON3 = sys.version_info >= (3, 0)

if IS_PYTHON3:
    basestring = str

    if sys.version_info >= (3, 4):
        from importlib.machinery import PathFinder


@plugin
class ScriptHost(object):

    """Provides an environment for running python plugins created for Vim."""

    def __init__(self, nvim):
        """Initialize the legacy python-vim environment."""
        self.setup(nvim)
        # context where all code will run
        self.module = imp.new_module('__main__')
        nvim.script_context = self.module
        # it seems some plugins assume 'sys' is already imported, so do it now
        exec('import sys', self.module.__dict__)

        self.legacy_vim = LegacyVim.from_nvim(nvim)
        sys.modules['vim'] = self.legacy_vim

    def setup(self, nvim):
        """Setup import hooks and global streams.

        This will add import hooks for importing modules from runtime
        directories and patch the sys module so 'print' calls will be
        forwarded to Nvim.
        """
        self.nvim = nvim
        info('install import hook/path')
        self.hook = path_hook(nvim)
        sys.path_hooks.append(self.hook)
        nvim.VIM_SPECIAL_PATH = '_vim_path_'
        sys.path.append(nvim.VIM_SPECIAL_PATH)
        info('redirect sys.stdout and sys.stderr')
        self.saved_stdout = sys.stdout
        self.saved_stderr = sys.stderr
        sys.stdout = RedirectStream(lambda data: nvim.out_write(data))

        sys.stderr = RedirectStream(lambda data: nvim.err_write(data))


    def teardown(self):
        """Restore state modified from the `setup` call."""
        nvim = self.nvim
        info('uninstall import hook/path')
        sys.path.remove(nvim.VIM_SPECIAL_PATH)
        sys.path_hooks.remove(self.hook)
        info('restore sys.stdout and sys.stderr')
        sys.stdout = self.saved_stdout
        sys.stderr = self.saved_stderr

    @rpc_export('python_execute', sync=True)
    def python_execute(self, script, range_start, range_stop):
        """Handle the `python` ex command."""
        self._set_current_range(range_start, range_stop)
        try:
            exec(script, self.module.__dict__)

        except Exception:
            raise ErrorResponse(format_exc_skip(1))

    @rpc_export('python_execute_async', sync=False)
    def python_execute_async(self, *args):
        try:
            self.python_execute(*args)
        except Exception:

            exception('python_execute failed')
            pass


    @rpc_export('python_execute_file', sync=True)
    def python_execute_file(self, file_path, range_start, range_stop):
        """Handle the `pyfile` ex command."""
        self._set_current_range(range_start, range_stop)
        with open(file_path) as f:
            script = compile(f.read(), file_path, 'exec')
            try:
                exec(script, self.module.__dict__)

            except Exception:
                raise ErrorResponse(format_exc_skip(1))

    @rpc_export('python_execute_file_async', sync=False)
    def python_execute_file_async(self, *args):
        try:
            self.python_execute(*args)
        except Exception as ex:

            exception('python_execute_file failed')
            pass


    @rpc_export('python_do_range', sync=True)
    def python_do_range(self, start, stop, code):
        """Handle the `pydo` ex command."""
        self._set_current_range(start, stop)
        nvim = self.nvim
        start -= 1
        fname = '_vim_pydo'

        # define the function
        function_def = 'def %s(line, linenr):\n %s' % (fname, code,)
        exec(function_def, self.module.__dict__)

        # get the function
        function = self.module.__dict__[fname]
        while start < stop:
            # Process batches of 5000 to avoid the overhead of making multiple
            # API calls for every line. Assuming an average line length of 100
            # bytes, approximately 488 kilobytes will be transferred per batch,
            # which can be done very quickly in a single API call.
            sstart = start
            sstop = min(start + 5000, stop)
            lines = nvim.current.buffer.api.get_lines(sstart, sstop, True)

            exception = None
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
            newlines = []
            linenr = sstart + 1
            for i, line in enumerate(lines):

                result = function(line, linenr)
                if result is None:
                    # Update earlier lines, and skip to the next
                    if newlines:
                        end = sstart + len(newlines) - 1
                        nvim.current.buffer.api.set_lines(sstart, end,
                                                          True, newlines)
                    sstart += len(newlines) + 1
                    newlines = []
                    pass

                elif isinstance(result, basestring):
                    newlines.append(result)
                else:
                    exception = TypeError('pydo should return a string ' +
                                          'or None, found %s instead'
                                          % result.__class__.__name__)
                    break
                linenr += 1

            start = sstop
            if newlines:
                end = sstart + len(newlines)
                nvim.current.buffer.api.set_lines(sstart, end, True, newlines)
            if exception:
                raise exception

        # delete the function
        del self.module.__dict__[fname]

    @rpc_export('python_eval', sync=True)
    def python_eval(self, expr):
        """Handle the `pyeval` vim function."""
        return eval(expr, self.module.__dict__)


    def _set_current_range(self, start, stop):
        current = self.legacy_vim.current
        current.range = current.buffer.range(start, stop)


class RedirectStream(io.IOBase):
    def __init__(self, redirect_handler):
        self.redirect_handler = redirect_handler

    def write(self, data):
        self.redirect_handler(data)

    def writelines(self, seq):
        self.redirect_handler('\n'.join(seq))


if IS_PYTHON3:
    num_types = (int, float)
else:
    num_types = (int, long, float)



def num_to_str(obj):
    if isinstance(obj, num_types):
        return str(obj)
    else:
        return obj


class LegacyVim(Nvim):
    def eval(self, expr):
        obj = self.request("vim_eval", expr)
        return walk(num_to_str, obj)


# This was copied/adapted from nvim-python help
def path_hook(nvim):
    def _get_paths():
        return discover_runtime_directories(nvim)

    def _find_module(fullname, oldtail, path):
        idx = oldtail.find('.')
        if idx > 0:
            name = oldtail[:idx]
            tail = oldtail[idx + 1:]
            fmr = imp.find_module(name, path)
            module = imp.find_module(fullname[:-len(oldtail)] + name, *fmr)
            return _find_module(fullname, tail, module.__path__)

        else:
            return imp.find_module(fullname, path)

    class VimModuleLoader(object):
        def __init__(self, module):
            self.module = module

        def load_module(self, fullname, path=None):

            # Check sys.modules, required for reload (see PEP302).
            if fullname in sys.modules:
                return sys.modules[fullname]
            return imp.load_module(fullname, *self.module)

    class VimPathFinder(object):
        @staticmethod
        def find_module(fullname, path=None):
            """Method for Python 2.7 and 3.3."""
            try:
                return VimModuleLoader(
                    _find_module(fullname, fullname, path or _get_paths()))
            except ImportError:
                return None

        @staticmethod
        def find_spec(fullname, path=None, target=None):
            """Method for Python 3.4+."""
            return PathFinder.find_spec(fullname, path or _get_paths(), target)

    def hook(path):
        if path == nvim.VIM_SPECIAL_PATH:
            return VimPathFinder
        else:
            raise ImportError

    return hook


def discover_runtime_directories(nvim):
    rv = []
    for path in nvim.list_runtime_paths():
        if not os.path.exists(path):
            continue
        path1 = os.path.join(path, 'pythonx')
        if IS_PYTHON3:
            path2 = os.path.join(path, 'python3')
        else:
            path2 = os.path.join(path, 'python2')
        if os.path.exists(path1):
            rv.append(path1)
        if os.path.exists(path2):
            rv.append(path2)
    return rv


1		"""Legacy python/python3-vim emulation."""
2	5	import imp
3	5	import io
4	5	import logging
5	5	import os
6	5	import sys
7
8	5	from .decorators import plugin, rpc_export
9	5	from ..api import Nvim, walk
10	5	from ..msgpack_rpc import ErrorResponse
11	5	from ..util import format_exc_skip
12
13	5	__all__ = ('ScriptHost',)
14
15
16	5	logger = logging.getLogger(__name__)
17	5	debug, info, warn, exception = (logger.debug,
18		logger.info,
19	5	logger.warn,
20		logger.exception)
21	5
22	3	IS_PYTHON3 = sys.version_info >= (3, 0)
23
24	3	if IS_PYTHON3:
25	2	basestring = str
26
27		if sys.version_info >= (3, 4):
28	5	from importlib.machinery import PathFinder
29	5
30
31		@plugin
32		class ScriptHost(object):
33	5
34		"""Provides an environment for running python plugins created for Vim."""
35
36		def __init__(self, nvim):
37		"""Initialize the legacy python-vim environment."""
38		self.setup(nvim)
39		# context where all code will run
40		self.module = imp.new_module('__main__')
41		nvim.script_context = self.module
42		# it seems some plugins assume 'sys' is already imported, so do it now
43		exec('import sys', self.module.__dict__)
		0 ignored issues – show Coding Style Security introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report The use of `exec` is discouraged. Execution of dynamic code might introduce security vulnerabilities. It is generally recommended to use this feature with care and only when necessary. Loading history...
44	5	self.legacy_vim = LegacyVim.from_nvim(nvim)
45		sys.modules['vim'] = self.legacy_vim
46
47		def setup(self, nvim):
48		"""Setup import hooks and global streams.
49
50		This will add import hooks for importing modules from runtime
51		directories and patch the sys module so 'print' calls will be
52		forwarded to Nvim.
53		"""
54		self.nvim = nvim
55		info('install import hook/path')
56		self.hook = path_hook(nvim)
57		sys.path_hooks.append(self.hook)
58		nvim.VIM_SPECIAL_PATH = '_vim_path_'
59		sys.path.append(nvim.VIM_SPECIAL_PATH)
60		info('redirect sys.stdout and sys.stderr')
61		self.saved_stdout = sys.stdout
62		self.saved_stderr = sys.stderr
63	5	sys.stdout = RedirectStream(lambda data: nvim.out_write(data))
		0 ignored issues – show Unused Code introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report This lambda might be unnecessary. Loading history...
64		sys.stderr = RedirectStream(lambda data: nvim.err_write(data))
		0 ignored issues – show Unused Code introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report This lambda might be unnecessary. Loading history...
65
66		def teardown(self):
67		"""Restore state modified from the `setup` call."""
68		nvim = self.nvim
69		info('uninstall import hook/path')
70		sys.path.remove(nvim.VIM_SPECIAL_PATH)
71		sys.path_hooks.remove(self.hook)
72		info('restore sys.stdout and sys.stderr')
73	5	sys.stdout = self.saved_stdout
74		sys.stderr = self.saved_stderr
75
76		@rpc_export('python_execute', sync=True)
77		def python_execute(self, script, range_start, range_stop):
78		"""Handle the `python` ex command."""
79		self._set_current_range(range_start, range_stop)
80		try:
81		exec(script, self.module.__dict__)
		0 ignored issues – show Coding Style Security introduced 2017-04-16 08:05 UTC by Report Bug Copy Issue Report The use of `exec` is discouraged. Execution of dynamic code might introduce security vulnerabilities. It is generally recommended to use this feature with care and only when necessary. Loading history...
82	5	except Exception:
83		raise ErrorResponse(format_exc_skip(1))
84
85		@rpc_export('python_execute_async', sync=False)
86		def python_execute_async(self, *args):
87		try:
88		self.python_execute(*args)
89		except Exception:
		0 ignored issues – show Best Practice introduced 2017-10-13 02:59 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...
90		exception('python_execute failed')
91		pass
		0 ignored issues – show Unused Code introduced 2017-10-13 02:59 UTC by Report Bug Copy Issue Report Unnecessary pass statement Loading history...
92
93	5	@rpc_export('python_execute_file', sync=True)
94		def python_execute_file(self, file_path, range_start, range_stop):
95		"""Handle the `pyfile` ex command."""
96		self._set_current_range(range_start, range_stop)
97		with open(file_path) as f:
98		script = compile(f.read(), file_path, 'exec')
99		try:
100		exec(script, self.module.__dict__)
		0 ignored issues – show Coding Style Security introduced 2016-06-08 19:50 UTC by Report Bug Copy Issue Report The use of `exec` is discouraged. Execution of dynamic code might introduce security vulnerabilities. It is generally recommended to use this feature with care and only when necessary. Loading history...
101		except Exception:
102		raise ErrorResponse(format_exc_skip(1))
103
104		@rpc_export('python_execute_file_async', sync=False)
105		def python_execute_file_async(self, *args):
106		try:
107		self.python_execute(*args)
108		except Exception as ex:
		0 ignored issues – show Best Practice introduced 2017-10-13 02:59 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... Unused Code introduced 2017-10-13 02:59 UTC by Report Bug Copy Issue Report The variable `ex` seems to be unused. Loading history...
109		exception('python_execute_file failed')
110		pass
		0 ignored issues – show Unused Code introduced 2017-10-13 02:59 UTC by Report Bug Copy Issue Report Unnecessary pass statement Loading history...
111
112		@rpc_export('python_do_range', sync=True)
113		def python_do_range(self, start, stop, code):
114		"""Handle the `pydo` ex command."""
115		self._set_current_range(start, stop)
116		nvim = self.nvim
117		start -= 1
118		fname = '_vim_pydo'
119
120		# define the function
121		function_def = 'def %s(line, linenr):\n %s' % (fname, code,)
122		exec(function_def, self.module.__dict__)
		0 ignored issues – show Coding Style Security introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report The use of `exec` is discouraged. Execution of dynamic code might introduce security vulnerabilities. It is generally recommended to use this feature with care and only when necessary. Loading history...
123		# get the function
124		function = self.module.__dict__[fname]
125		while start < stop:
126		# Process batches of 5000 to avoid the overhead of making multiple
127		# API calls for every line. Assuming an average line length of 100
128		# bytes, approximately 488 kilobytes will be transferred per batch,
129		# which can be done very quickly in a single API call.
130		sstart = start
131		sstop = min(start + 5000, stop)
132		lines = nvim.current.buffer.api.get_lines(sstart, sstop, True)
133
134		exception = None
		0 ignored issues – show Comprehensibility Bug introduced 2017-10-13 02:59 UTC by Report Bug Copy Issue Report `exception` is re-defining a name which is already available in the outer-scope (previously defined on line `17`). It is generally a bad practice to shadow variables from the outer-scope. In most cases, this is done unintentionally and might lead to unexpected behavior: param = 5 class Foo: def __init__(self, param): # "param" would be flagged here self.param = param Loading history...
135		newlines = []
136		linenr = sstart + 1
137		for i, line in enumerate(lines):
		0 ignored issues – show Unused Code introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report The variable `i` seems to be unused. Loading history...
138		result = function(line, linenr)
139		if result is None:
140		# Update earlier lines, and skip to the next
141		if newlines:
142		end = sstart + len(newlines) - 1
143		nvim.current.buffer.api.set_lines(sstart, end,
144		True, newlines)
145		sstart += len(newlines) + 1
146		newlines = []
147	5	pass
		0 ignored issues – show Unused Code introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report Unnecessary pass statement Loading history...
148		elif isinstance(result, basestring):
149		newlines.append(result)
150		else:
151		exception = TypeError('pydo should return a string ' +
152	5	'or None, found %s instead'
153		% result.__class__.__name__)
154		break
155		linenr += 1
156
157	5	start = sstop
158	5	if newlines:
159		end = sstart + len(newlines)
160		nvim.current.buffer.api.set_lines(sstart, end, True, newlines)
161	5	if exception:
162		raise exception
		0 ignored issues – show Bug introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report Raising NoneType while only classes or instances are allowed Loading history...
163		# delete the function
164	5	del self.module.__dict__[fname]
165
166		@rpc_export('python_eval', sync=True)
167		def python_eval(self, expr):
168	5	"""Handle the `pyeval` vim function."""
169	3	return eval(expr, self.module.__dict__)
		0 ignored issues – show Security Best Practice introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report `eval` should generally only be used if absolutely necessary. The usage of `eval` might allow for executing arbitrary code if a user manages to inject dynamic input. Please use this language feature with care and only when you are sure of the input. Loading history...
170
171	2	def _set_current_range(self, start, stop):
172		current = self.legacy_vim.current
173		current.range = current.buffer.range(start, stop)
174	5
175
176		class RedirectStream(io.IOBase):
177		def __init__(self, redirect_handler):
178		self.redirect_handler = redirect_handler
179
180		def write(self, data):
181	5	self.redirect_handler(data)
182	5
183		def writelines(self, seq):
184		self.redirect_handler('\n'.join(seq))
185
186
187		if IS_PYTHON3:
188	5	num_types = (int, float)
189		else:
190		num_types = (int, long, float)
		0 ignored issues – show Comprehensibility Best Practice introduced 2016-04-08 13:03 UTC by Report Bug Copy Issue Report Undefined variable 'long' Loading history...
191
192
193		def num_to_str(obj):
194		if isinstance(obj, num_types):
195		return str(obj)
196		else:
197		return obj
198
199
200		class LegacyVim(Nvim):
201		def eval(self, expr):
202		obj = self.request("vim_eval", expr)
203		return walk(num_to_str, obj)
204
205
206		# This was copied/adapted from nvim-python help
207		def path_hook(nvim):
208		def _get_paths():
209		return discover_runtime_directories(nvim)
210
211		def _find_module(fullname, oldtail, path):
212		idx = oldtail.find('.')
213		if idx > 0:
214		name = oldtail[:idx]
215		tail = oldtail[idx + 1:]
216		fmr = imp.find_module(name, path)
217		module = imp.find_module(fullname[:-len(oldtail)] + name, *fmr)
218		return _find_module(fullname, tail, module.__path__)
		0 ignored issues – show Bug introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report The Instance of `tuple` does not seem to have a member named `__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...
219		else:
220		return imp.find_module(fullname, path)
221
222		class VimModuleLoader(object):
223		def __init__(self, module):
224		self.module = module
225
226		def load_module(self, fullname, path=None):
		0 ignored issues – show Unused Code introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report The argument `path` seems to be unused. Loading history...
227		# Check sys.modules, required for reload (see PEP302).
228		if fullname in sys.modules:
229		return sys.modules[fullname]
230		return imp.load_module(fullname, *self.module)
231
232		class VimPathFinder(object):
233		@staticmethod
234		def find_module(fullname, path=None):
235		"""Method for Python 2.7 and 3.3."""
236		try:
237	5	return VimModuleLoader(
238		_find_module(fullname, fullname, path or _get_paths()))
239		except ImportError:
240		return None
241
242		@staticmethod
243		def find_spec(fullname, path=None, target=None):
244		"""Method for Python 3.4+."""
245		return PathFinder.find_spec(fullname, path or _get_paths(), target)
246
247		def hook(path):
248		if path == nvim.VIM_SPECIAL_PATH:
249		return VimPathFinder
250		else:
251		raise ImportError
252
253		return hook
254
255
256		def discover_runtime_directories(nvim):
257		rv = []
258		for path in nvim.list_runtime_paths():
259		if not os.path.exists(path):
260		continue
261		path1 = os.path.join(path, 'pythonx')
262		if IS_PYTHON3:
263		path2 = os.path.join(path, 'python3')
264		else:
265		path2 = os.path.join(path, 'python2')
266		if os.path.exists(path1):
267		rv.append(path1)
268		if os.path.exists(path2):
269		rv.append(path2)
270		return rv
271

neovim / python-client

Pull Request — master (#273)

ScriptHost.python_execute_async() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like