VimModuleLoader.load_module() - Code Metrics - Inspection of "change default unicode error mode to "surrogateesc..." - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#199)

by Björn

created 2016-04-21 19:26 UTC

VimModuleLoader.load_module() A

↳ Parent: VimModuleLoader

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Metric	Value
cc	2
dl	0
loc	5
ccs	0
cts	3
cp	0
crap	6
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

__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


@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)
        exec(script, self.module.__dict__)


    @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')
            exec(script, self.module.__dict__)


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

neovim / python-client

Pull Request — master (#199)

VimModuleLoader.load_module() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like