ScriptHost.python_execute_file() - Code Metrics - Inspection of "set process-global CWD on DirChanged #296" - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 70dfc8...1ab98e )

by Justin M.

created 2018-04-13 00:09 UTC

ScriptHost.python_execute_file() A

↳ Parent: ScriptHost

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	3
c	2
b	0
f	0
dl	0
loc	10
ccs	0
cts	6
cp	0
crap	12
rs	9.4285

"""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 = (logger.debug, logger.info, logger.warn,)

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

if IS_PYTHON3:
    basestring = str

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

    PYTHON_SUBDIR = 'python3'
else:
    PYTHON_SUBDIR = 'python2'


@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

        # Handle DirChanged. #296
        nvim.command(
            'autocmd DirChanged * call rpcrequest({}, "python_chdir", v:event)'
            .format(nvim.channel_id), async=True)
        # XXX: Avoid race condition.
        # https://github.com/neovim/python-client/pull/296#issuecomment-358970531
        os.chdir(nvim.eval('getcwd()', async=False))

    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_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_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
            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__)


    @rpc_export('python_chdir', sync=True)
    def python_chdir(self, args):
class Foo:
    def some_method(self, x, y):
        return x + y;
        """Handle working directory changes."""
        os.chdir(args['cwd'])

    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)  # noqa: F821



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 rtp in nvim.list_runtime_paths():
        if not os.path.exists(rtp):
            continue
        for subdir in ['pythonx', PYTHON_SUBDIR]:
            path = os.path.join(rtp, subdir)
            if os.path.exists(path):
                rv.append(path)
    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 = (logger.debug, logger.info, logger.warn,)
18
19	5	IS_PYTHON3 = sys.version_info >= (3, 0)
20
21	5	if IS_PYTHON3:
22	3	basestring = str
23
24	3	if sys.version_info >= (3, 4):
25	2	from importlib.machinery import PathFinder
26
27		PYTHON_SUBDIR = 'python3'
28	5	else:
29	5	PYTHON_SUBDIR = 'python2'
30
31
32		@plugin
33	5	class ScriptHost(object):
34
35		"""Provides an environment for running python plugins created for Vim."""
36
37		def __init__(self, nvim):
38		"""Initialize the legacy python-vim environment."""
39		self.setup(nvim)
40		# context where all code will run
41		self.module = imp.new_module('__main__')
42		nvim.script_context = self.module
43		# it seems some plugins assume 'sys' is already imported, so do it now
44	5	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...
45		self.legacy_vim = LegacyVim.from_nvim(nvim)
46		sys.modules['vim'] = self.legacy_vim
47
48		# Handle DirChanged. #296
49		nvim.command(
50		'autocmd DirChanged * call rpcrequest({}, "python_chdir", v:event)'
51		.format(nvim.channel_id), async=True)
52		# XXX: Avoid race condition.
53		# https://github.com/neovim/python-client/pull/296#issuecomment-358970531
54		os.chdir(nvim.eval('getcwd()', async=False))
55
56		def setup(self, nvim):
57		"""Setup import hooks and global streams.
58
59		This will add import hooks for importing modules from runtime
60		directories and patch the sys module so 'print' calls will be
61		forwarded to Nvim.
62		"""
63	5	self.nvim = nvim
64		info('install import hook/path')
65		self.hook = path_hook(nvim)
66		sys.path_hooks.append(self.hook)
67		nvim.VIM_SPECIAL_PATH = '_vim_path_'
68		sys.path.append(nvim.VIM_SPECIAL_PATH)
69		info('redirect sys.stdout and sys.stderr')
70		self.saved_stdout = sys.stdout
71		self.saved_stderr = sys.stderr
72		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...
73	5	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...
74
75		def teardown(self):
76		"""Restore state modified from the `setup` call."""
77		nvim = self.nvim
78		info('uninstall import hook/path')
79		sys.path.remove(nvim.VIM_SPECIAL_PATH)
80		sys.path_hooks.remove(self.hook)
81		info('restore sys.stdout and sys.stderr')
82	5	sys.stdout = self.saved_stdout
83		sys.stderr = self.saved_stderr
84
85		@rpc_export('python_execute', sync=True)
86		def python_execute(self, script, range_start, range_stop):
87		"""Handle the `python` ex command."""
88		self._set_current_range(range_start, range_stop)
89		try:
90		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...
91		except Exception:
92		raise ErrorResponse(format_exc_skip(1))
93	5
94		@rpc_export('python_execute_file', sync=True)
95		def python_execute_file(self, file_path, range_start, range_stop):
96		"""Handle the `pyfile` ex command."""
97		self._set_current_range(range_start, range_stop)
98		with open(file_path) as f:
99		script = compile(f.read(), file_path, 'exec')
100		try:
101		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...
102		except Exception:
103		raise ErrorResponse(format_exc_skip(1))
104
105		@rpc_export('python_do_range', sync=True)
106		def python_do_range(self, start, stop, code):
107		"""Handle the `pydo` ex command."""
108		self._set_current_range(start, stop)
109		nvim = self.nvim
110		start -= 1
111		fname = '_vim_pydo'
112
113		# define the function
114		function_def = 'def %s(line, linenr):\n %s' % (fname, code,)
115		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...
116		# get the function
117		function = self.module.__dict__[fname]
118		while start < stop:
119		# Process batches of 5000 to avoid the overhead of making multiple
120		# API calls for every line. Assuming an average line length of 100
121		# bytes, approximately 488 kilobytes will be transferred per batch,
122		# which can be done very quickly in a single API call.
123		sstart = start
124		sstop = min(start + 5000, stop)
125		lines = nvim.current.buffer.api.get_lines(sstart, sstop, True)
126
127		exception = None
128		newlines = []
129		linenr = sstart + 1
130		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...
131		result = function(line, linenr)
132		if result is None:
133		# Update earlier lines, and skip to the next
134		if newlines:
135		end = sstart + len(newlines) - 1
136		nvim.current.buffer.api.set_lines(sstart, end,
137		True, newlines)
138		sstart += len(newlines) + 1
139		newlines = []
140		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...
141		elif isinstance(result, basestring):
142		newlines.append(result)
143		else:
144		exception = TypeError('pydo should return a string ' +
145		'or None, found %s instead'
146		% result.__class__.__name__)
147	5	break
148		linenr += 1
149
150		start = sstop
151		if newlines:
152	5	end = sstart + len(newlines)
153		nvim.current.buffer.api.set_lines(sstart, end, True, newlines)
154		if exception:
155		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...
156		# delete the function
157	5	del self.module.__dict__[fname]
158	5
159		@rpc_export('python_eval', sync=True)
160		def python_eval(self, expr):
161	5	"""Handle the `pyeval` vim function."""
162		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...
163
164	5	@rpc_export('python_chdir', sync=True)
165		def python_chdir(self, args):
		0 ignored issues – show Coding Style introduced 2018-01-19 11:18 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...
166		"""Handle working directory changes."""
167		os.chdir(args['cwd'])
168	5
169	3	def _set_current_range(self, start, stop):
170		current = self.legacy_vim.current
171	2	current.range = current.buffer.range(start, stop)
172
173
174	5	class RedirectStream(io.IOBase):
175		def __init__(self, redirect_handler):
176		self.redirect_handler = redirect_handler
177
178		def write(self, data):
179		self.redirect_handler(data)
180
181	5	def writelines(self, seq):
182	5	self.redirect_handler('\n'.join(seq))
183
184
185		if IS_PYTHON3:
186		num_types = (int, float)
187		else:
188	5	num_types = (int, long, float) # noqa: F821
		0 ignored issues – show Comprehensibility Best Practice introduced 2017-12-30 15:38 UTC by Report Bug Copy Issue Report Undefined variable 'long' Loading history...
189
190
191		def num_to_str(obj):
192		if isinstance(obj, num_types):
193		return str(obj)
194		else:
195		return obj
196
197
198		class LegacyVim(Nvim):
199		def eval(self, expr):
200		obj = self.request("vim_eval", expr)
201		return walk(num_to_str, obj)
202
203
204		# This was copied/adapted from nvim-python help
205		def path_hook(nvim):
206		def _get_paths():
207		return discover_runtime_directories(nvim)
208
209		def _find_module(fullname, oldtail, path):
210		idx = oldtail.find('.')
211		if idx > 0:
212		name = oldtail[:idx]
213		tail = oldtail[idx + 1:]
214		fmr = imp.find_module(name, path)
215		module = imp.find_module(fullname[:-len(oldtail)] + name, *fmr)
216		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...
217		else:
218		return imp.find_module(fullname, path)
219
220		class VimModuleLoader(object):
221		def __init__(self, module):
222		self.module = module
223
224		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...
225		# Check sys.modules, required for reload (see PEP302).
226		if fullname in sys.modules:
227		return sys.modules[fullname]
228		return imp.load_module(fullname, *self.module)
229
230		class VimPathFinder(object):
231		@staticmethod
232		def find_module(fullname, path=None):
233		"""Method for Python 2.7 and 3.3."""
234		try:
235		return VimModuleLoader(
236		_find_module(fullname, fullname, path or _get_paths()))
237	5	except ImportError:
238		return None
239
240		@staticmethod
241		def find_spec(fullname, path=None, target=None):
242		"""Method for Python 3.4+."""
243		return PathFinder.find_spec(fullname, path or _get_paths(), target)
244
245		def hook(path):
246		if path == nvim.VIM_SPECIAL_PATH:
247		return VimPathFinder
248		else:
249		raise ImportError
250
251		return hook
252
253
254		def discover_runtime_directories(nvim):
255		rv = []
256		for rtp in nvim.list_runtime_paths():
257		if not os.path.exists(rtp):
258		continue
259		for subdir in ['pythonx', PYTHON_SUBDIR]:
260		path = os.path.join(rtp, subdir)
261		if os.path.exists(path):
262		rv.append(path)
263		return rv
264

neovim / python-client

Push — master ( 70dfc8...1ab98e )

ScriptHost.python_execute_file() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like