src.hunter.ColorStreamAction.stream() - Code Metrics - Inspection of "Bump version: 1.2.1 → 1.2.2" - ionelmc/python-hunter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 70f629...0878c0 )

by Ionel Cristian

created 2016-01-30 09:41 UTC

src.hunter.ColorStreamAction.stream() B

↳ Parent: src.hunter.ColorStreamAction

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	19
rs	7.3333
cc	7

from __future__ import absolute_import

import ast
import os
import pdb
import sys
from itertools import chain

from colorama import AnsiToWin32
from colorama import Back
from colorama import Fore
from colorama import Style
from six import string_types

from .util import Fields


EVENT_COLORS = {
    'reset': Style.RESET_ALL,
    'normal': Style.NORMAL,
    'filename': '',
    'colon': Style.BRIGHT + Fore.BLACK,
    'lineno': Style.RESET_ALL,
    'kind': Fore.CYAN,
    'continuation': Style.BRIGHT + Fore.BLUE,
    'call': Style.BRIGHT + Fore.BLUE,
    'return': Style.BRIGHT + Fore.GREEN,
    'exception': Style.BRIGHT + Fore.RED,
    'detail': Style.NORMAL,
    'vars': Style.RESET_ALL + Fore.MAGENTA,
    'vars-name': Style.BRIGHT,
    'internal-failure': Style.BRIGHT + Back.RED + Fore.RED,
    'internal-detail': Fore.WHITE,
    'source-failure': Style.BRIGHT + Back.YELLOW + Fore.YELLOW,
    'source-detail': Fore.WHITE,
}
CODE_COLORS = {
    'call': Fore.RESET + Style.BRIGHT,
    'line': Fore.RESET,
    'return': Fore.YELLOW,
    'exception': Fore.RED,
}
NO_COLORS = {key: '' for key in chain(CODE_COLORS, EVENT_COLORS)}
MISSING = type('MISSING', (), {'__repr__': lambda _: '?'})()


class Action(object):
    def __call__(self, event):
        raise NotImplementedError()


class Debugger(Fields.klass.kwargs, Action):
    """
    An action that starts ``pdb``.
    """

    def __init__(self, klass=pdb.Pdb, **kwargs):
        self.klass = klass
        self.kwargs = kwargs

    def __call__(self, event):
        """
        Runs a ``pdb.set_trace`` at the matching frame.
        """
        self.klass(**self.kwargs).set_trace(event.frame)


class ColorStreamAction(Fields.stream.force_colors.filename_alignment.repr_limit, Action):
    _stream_cache = {}
    _stream = None
    _tty = None

    def __init__(self,
                 stream=sys.stderr,
                 force_colors=False,
                 filename_alignment=40,
                 repr_limit=1024):
        self.force_colors = force_colors
        self.stream = stream
        self.filename_alignment = max(5, filename_alignment)
        self.repr_limit = repr_limit

    @property
    def stream(self):
        return self._stream

    @stream.setter
    def stream(self, value):
        if isinstance(value, string_types):
            if value in self._stream_cache:
                value = self._stream_cache[value]
            else:
                value = self._stream_cache[value] = open(value, 'a', buffering=0)

        isatty = getattr(value, 'isatty', None)
        if self.force_colors or (isatty and isatty() and os.name != 'java'):
            self._stream = AnsiToWin32(value, strip=False)
            self._tty = True
            self.event_colors = EVENT_COLORS
            self.code_colors = CODE_COLORS
        else:
            self._tty = False
            self._stream = value
            self.event_colors = NO_COLORS
            self.code_colors = NO_COLORS

    def _safe_repr(self, obj):
        limit = self.repr_limit

        try:
            s = repr(obj)
            s = s.replace('\n', r'\n')
            if len(s) > limit:
                cutoff = limit // 2
                return "{} {continuation}[...]{reset} {}".format(s[:cutoff], s[-cutoff:], **self.event_colors)
            else:
                return s
        except Exception as exc:
            return "{internal-failure}!!! FAILED REPR: {internal-detail}{!r}{reset}".format(exc, **self.event_colors)


class CodePrinter(ColorStreamAction):
    """
    An action that just prints the code being executed.

    Args:
        stream (file-like): Stream to write to. Default: ``sys.stderr``.
        filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
        force_colors (bool): Force coloring. Default: ``False``.
        repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.
    """
    def _safe_source(self, event):
        try:
            lines = event._raw_fullsource.rstrip().splitlines()
            if lines:
                return lines
            else:
                return "{source-failure}??? NO SOURCE: {source-detail}" \
                       "Source code string for module {!r} is empty.".format(event.module, **self.event_colors),
            return lines
        except Exception as exc:
            return "{source-failure}??? NO SOURCE: {source-detail}{!r}".format(exc, **self.event_colors),

    def _format_filename(self, event):
        filename = event.filename or "<???>"
        if len(filename) > self.filename_alignment:
            filename = '[...]{}'.format(filename[5 - self.filename_alignment:])
        return filename

    def __call__(self, event, sep=os.path.sep, join=os.path.join):
        """
        Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
        prints values.
        """

        # context = event.tracer
        # alignment = context.filename_alignment = max(
        #     getattr(context, 'filename_alignment', 5),
        #     len(filename)
        # )
        lines = self._safe_source(event)
        self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{reset}\n".format(
            self._format_filename(event),
            event.lineno,
            event.kind,
            lines[0],
            align=self.filename_alignment,
            code=self.code_colors[event.kind],
            **self.event_colors
        ))
        for line in lines[1:]:
            self.stream.write("{:>{align}}       {kind}{:9} {code}{}{reset}\n".format(
                "",
                r"   |",
                line,
                align=self.filename_alignment,
                code=self.code_colors[event.kind],
                **self.event_colors
            ))

        if event.kind in ('return', 'exception'):
            self.stream.write("{:>{align}}       {continuation}{:9} {color}{} value: {detail}{}{reset}\n".format(
                "",
                "...",
                event.kind,
                self._safe_repr(event.arg),
                align=self.filename_alignment,
                color=self.event_colors[event.kind],
                **self.event_colors
            ))


class CallPrinter(CodePrinter):
    """
    An action that just prints the code being executed, but unlike :obj:`hunter.CodePrinter` it indents based on
    callstack depth and it also shows ``repr()`` of function arguments.

    Args:
        stream (file-like): Stream to write to. Default: ``sys.stderr``.
        filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
        force_colors (bool): Force coloring. Default: ``False``.
        repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.

    .. versionadded:: 1.2.0

    .. note::

        This will be the default action in `hunter 2.0`.
    """

    def __init__(self, **options):
        super(CallPrinter, self).__init__(**options)
        self.stack = []

    def __call__(self, event, sep=os.path.sep, join=os.path.join):
        """
        Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
        prints values.
        """
        filename = self._format_filename(event)
        ident = event.module, event.function

        if event.kind == 'call':
            code = event.code
            self.stack.append(ident)
            self.stream.write(
                "{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {}{call}=>{normal} "
                "{}({}{call}{normal}){reset}\n".format(
                    filename,
                    event.lineno,
                    event.kind,
                    '   ' * (len(self.stack) - 1),
                    event.function,
                    ', '.join('{vars}{vars-name}{0}{vars}={reset}{1}'.format(
                        var,
                        self._safe_repr(event.locals.get(var, MISSING)),
                        **self.event_colors
                    ) for var in code.co_varnames[:code.co_argcount]),
                    align=self.filename_alignment,
                    **self.event_colors
                ))
        elif event.kind in ('return', 'exception'):
            self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{}{normal} {}: {reset}{}\n".format(
                filename,
                event.lineno,
                event.kind,
                '   ' * (len(self.stack) - 1),
                {'return': '<=', 'exception': '<!'}[event.kind],
                event.function,
                self._safe_repr(event.arg),
                align=self.filename_alignment,
                code=self.event_colors[event.kind],
                **self.event_colors
            ))
            if self.stack and self.stack[-1] == ident:
                self.stack.pop()
        else:
            self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {reset}{}{}\n".format(
                filename,
                event.lineno,
                event.kind,
                '   ' * len(self.stack),
                event.source.strip(),
                align=self.filename_alignment,
                code=self.code_colors[event.kind],
                **self.event_colors
            ))


class VarsPrinter(Fields.names.globals.stream.filename_alignment, ColorStreamAction):
    """
    An action that prints local variables and optionally global variables visible from the current executing frame.

    Args:
        *names (strings): Names to evaluate. Expressions can be used (will only try to evaluate if all the variables are
            present on the frame.
        globals (bool): Allow access to globals. Default: ``False`` (only looks at locals).
        stream (file-like): Stream to write to. Default: ``sys.stderr``.
        filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
        force_colors (bool): Force coloring. Default: ``False``.
        repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.

    .. note::
        This is the default action.

    .. warning::

        In `hunter 2.0` the default action will be :obj:`hunter.CallPrinter`.
    """

    def __init__(self, *names, **options):
        if not names:
            raise TypeError("VarsPrinter requires at least one variable name/expression.")
        self.names = {
            name: set(self._iter_symbols(name))
            for name in names
        }
        self.globals = options.pop('globals', False)
        super(VarsPrinter, self).__init__(**options)

    @staticmethod
    def _iter_symbols(code):
        """
        Iterate all the variable names in the given expression.

        Example:

        * ``self.foobar`` yields ``self``
        * ``self[foobar]`` yields `self`` and ``foobar``
        """
        for node in ast.walk(ast.parse(code)):
            if isinstance(node, ast.Name):
                yield node.id

    def _safe_eval(self, code, event):
        """
        Try to evaluate the given code on the given frame. If failure occurs, returns some ugly string with exception.
        """
        try:
            return eval(code, event.globals if self.globals else {}, event.locals)
        except Exception as exc:
            return "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)

    def __call__(self, event):
        """
        Handle event and print the specified variables.
        """
        first = True
        frame_symbols = set(event.locals)
        if self.globals:
            frame_symbols |= set(event.globals)

        for code, symbols in self.names.items():
            try:
                obj = eval(code, event.globals if self.globals else {}, event.locals)
            except AttributeError:
                continue
            except Exception as exc:
                printout = "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
            else:
                printout = self._safe_repr(obj)

            if frame_symbols >= symbols:
                self.stream.write("{:>{align}}       {vars}{:9} {vars-name}{} {vars}=> {reset}{}{reset}\n".format(
                    "",
                    "vars" if first else "...",
                    code,
                    printout,
                    align=self.filename_alignment,
                    **self.event_colors
                ))
                first = False


1			from __future__ import absolute_import
2
3			import ast
4			import os
5			import pdb
6			import sys
7			from itertools import chain
8
9			from colorama import AnsiToWin32
10			from colorama import Back
11			from colorama import Fore
12			from colorama import Style
13			from six import string_types
14
15			from .util import Fields
16
17
18			EVENT_COLORS = {
19			'reset': Style.RESET_ALL,
20			'normal': Style.NORMAL,
21			'filename': '',
22			'colon': Style.BRIGHT + Fore.BLACK,
23			'lineno': Style.RESET_ALL,
24			'kind': Fore.CYAN,
25			'continuation': Style.BRIGHT + Fore.BLUE,
26			'call': Style.BRIGHT + Fore.BLUE,
27			'return': Style.BRIGHT + Fore.GREEN,
28			'exception': Style.BRIGHT + Fore.RED,
29			'detail': Style.NORMAL,
30			'vars': Style.RESET_ALL + Fore.MAGENTA,
31			'vars-name': Style.BRIGHT,
32			'internal-failure': Style.BRIGHT + Back.RED + Fore.RED,
33			'internal-detail': Fore.WHITE,
34			'source-failure': Style.BRIGHT + Back.YELLOW + Fore.YELLOW,
35			'source-detail': Fore.WHITE,
36			}
37			CODE_COLORS = {
38			'call': Fore.RESET + Style.BRIGHT,
39			'line': Fore.RESET,
40			'return': Fore.YELLOW,
41			'exception': Fore.RED,
42			}
43			NO_COLORS = {key: '' for key in chain(CODE_COLORS, EVENT_COLORS)}
44			MISSING = type('MISSING', (), {'__repr__': lambda _: '?'})()
45
46
47			class Action(object):
48			def __call__(self, event):
49			raise NotImplementedError()
50
51
52			class Debugger(Fields.klass.kwargs, Action):
53			"""
54			An action that starts ``pdb``.
55			"""
56
57			def __init__(self, klass=pdb.Pdb, **kwargs):
58			self.klass = klass
59			self.kwargs = kwargs
60
61			def __call__(self, event):
62			"""
63			Runs a ``pdb.set_trace`` at the matching frame.
64			"""
65			self.klass(**self.kwargs).set_trace(event.frame)
66
67
68			class ColorStreamAction(Fields.stream.force_colors.filename_alignment.repr_limit, Action):
69			_stream_cache = {}
70			_stream = None
71			_tty = None
72
73			def __init__(self,
74			stream=sys.stderr,
75			force_colors=False,
76			filename_alignment=40,
77			repr_limit=1024):
78			self.force_colors = force_colors
79			self.stream = stream
80			self.filename_alignment = max(5, filename_alignment)
81			self.repr_limit = repr_limit
82
83			@property
84			def stream(self):
85			return self._stream
86
87			@stream.setter
88			def stream(self, value):
89			if isinstance(value, string_types):
90			if value in self._stream_cache:
91			value = self._stream_cache[value]
92			else:
93			value = self._stream_cache[value] = open(value, 'a', buffering=0)
94
95			isatty = getattr(value, 'isatty', None)
96			if self.force_colors or (isatty and isatty() and os.name != 'java'):
97			self._stream = AnsiToWin32(value, strip=False)
98			self._tty = True
99			self.event_colors = EVENT_COLORS
100			self.code_colors = CODE_COLORS
101			else:
102			self._tty = False
103			self._stream = value
104			self.event_colors = NO_COLORS
105			self.code_colors = NO_COLORS
106
107			def _safe_repr(self, obj):
108			limit = self.repr_limit
109
110			try:
111			s = repr(obj)
112			s = s.replace('\n', r'\n')
113			if len(s) > limit:
114			cutoff = limit // 2
115			return "{} {continuation}[...]{reset} {}".format(s[:cutoff], s[-cutoff:], **self.event_colors)
116			else:
117			return s
118			except Exception as exc:
119			return "{internal-failure}!!! FAILED REPR: {internal-detail}{!r}{reset}".format(exc, **self.event_colors)
120
121
122			class CodePrinter(ColorStreamAction):
123			"""
124			An action that just prints the code being executed.
125
126			Args:
127			stream (file-like): Stream to write to. Default: ``sys.stderr``.
128			filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
129			force_colors (bool): Force coloring. Default: ``False``.
130			repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.
131			"""
132			def _safe_source(self, event):
133			try:
134			lines = event._raw_fullsource.rstrip().splitlines()
135			if lines:
136			return lines
137			else:
138			return "{source-failure}??? NO SOURCE: {source-detail}" \
139			"Source code string for module {!r} is empty.".format(event.module, **self.event_colors),
140			return lines
141			except Exception as exc:
142			return "{source-failure}??? NO SOURCE: {source-detail}{!r}".format(exc, **self.event_colors),
143
144			def _format_filename(self, event):
145			filename = event.filename or "<???>"
146			if len(filename) > self.filename_alignment:
147			filename = '[...]{}'.format(filename[5 - self.filename_alignment:])
148			return filename
149
150			def __call__(self, event, sep=os.path.sep, join=os.path.join):
151			"""
152			Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
153			prints values.
154			"""
155
156			# context = event.tracer
157			# alignment = context.filename_alignment = max(
158			# getattr(context, 'filename_alignment', 5),
159			# len(filename)
160			# )
161			lines = self._safe_source(event)
162			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{reset}\n".format(
163			self._format_filename(event),
164			event.lineno,
165			event.kind,
166			lines[0],
167			align=self.filename_alignment,
168			code=self.code_colors[event.kind],
169			**self.event_colors
170			))
171			for line in lines[1:]:
172			self.stream.write("{:>{align}} {kind}{:9} {code}{}{reset}\n".format(
173			"",
174			r" \|",
175			line,
176			align=self.filename_alignment,
177			code=self.code_colors[event.kind],
178			**self.event_colors
179			))
180
181			if event.kind in ('return', 'exception'):
182			self.stream.write("{:>{align}} {continuation}{:9} {color}{} value: {detail}{}{reset}\n".format(
183			"",
184			"...",
185			event.kind,
186			self._safe_repr(event.arg),
187			align=self.filename_alignment,
188			color=self.event_colors[event.kind],
189			**self.event_colors
190			))
191
192
193			class CallPrinter(CodePrinter):
194			"""
195			An action that just prints the code being executed, but unlike :obj:`hunter.CodePrinter` it indents based on
196			callstack depth and it also shows ``repr()`` of function arguments.
197
198			Args:
199			stream (file-like): Stream to write to. Default: ``sys.stderr``.
200			filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
201			force_colors (bool): Force coloring. Default: ``False``.
202			repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.
203
204			.. versionadded:: 1.2.0
205
206			.. note::
207
208			This will be the default action in `hunter 2.0`.
209			"""
210
211			def __init__(self, **options):
212			super(CallPrinter, self).__init__(**options)
213			self.stack = []
214
215			def __call__(self, event, sep=os.path.sep, join=os.path.join):
216			"""
217			Handle event and print filename, line number and source code. If event.kind is a `return` or `exception` also
218			prints values.
219			"""
220			filename = self._format_filename(event)
221			ident = event.module, event.function
222
223			if event.kind == 'call':
224			code = event.code
225			self.stack.append(ident)
226			self.stream.write(
227			"{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {}{call}=>{normal} "
228			"{}({}{call}{normal}){reset}\n".format(
229			filename,
230			event.lineno,
231			event.kind,
232			' ' * (len(self.stack) - 1),
233			event.function,
234			', '.join('{vars}{vars-name}{0}{vars}={reset}{1}'.format(
235			var,
236			self._safe_repr(event.locals.get(var, MISSING)),
237			**self.event_colors
238			) for var in code.co_varnames[:code.co_argcount]),
239			align=self.filename_alignment,
240			**self.event_colors
241			))
242			elif event.kind in ('return', 'exception'):
243			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{}{normal} {}: {reset}{}\n".format(
244			filename,
245			event.lineno,
246			event.kind,
247			' ' * (len(self.stack) - 1),
248			{'return': '<=', 'exception': '<!'}[event.kind],
249			event.function,
250			self._safe_repr(event.arg),
251			align=self.filename_alignment,
252			code=self.event_colors[event.kind],
253			**self.event_colors
254			))
255			if self.stack and self.stack[-1] == ident:
256			self.stack.pop()
257			else:
258			self.stream.write("{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {reset}{}{}\n".format(
259			filename,
260			event.lineno,
261			event.kind,
262			' ' * len(self.stack),
263			event.source.strip(),
264			align=self.filename_alignment,
265			code=self.code_colors[event.kind],
266			**self.event_colors
267			))
268
269
270			class VarsPrinter(Fields.names.globals.stream.filename_alignment, ColorStreamAction):
271			"""
272			An action that prints local variables and optionally global variables visible from the current executing frame.
273
274			Args:
275			*names (strings): Names to evaluate. Expressions can be used (will only try to evaluate if all the variables are
276			present on the frame.
277			globals (bool): Allow access to globals. Default: ``False`` (only looks at locals).
278			stream (file-like): Stream to write to. Default: ``sys.stderr``.
279			filename_alignment (int): Default size for the filename column (files are right-aligned). Default: ``40``.
280			force_colors (bool): Force coloring. Default: ``False``.
281			repr_limit (bool): Limit length of ``repr()`` output. Default: ``512``.
282
283			.. note::
284			This is the default action.
285
286			.. warning::
287
288			In `hunter 2.0` the default action will be :obj:`hunter.CallPrinter`.
289			"""
290
291			def __init__(self, names, *options):
292			if not names:
293			raise TypeError("VarsPrinter requires at least one variable name/expression.")
294			self.names = {
295			name: set(self._iter_symbols(name))
296			for name in names
297			}
298			self.globals = options.pop('globals', False)
299			super(VarsPrinter, self).__init__(**options)
300
301			@staticmethod
302			def _iter_symbols(code):
303			"""
304			Iterate all the variable names in the given expression.
305
306			Example:
307
308			* ``self.foobar`` yields ``self``
309			* ``self[foobar]`` yields `self`` and ``foobar``
310			"""
311			for node in ast.walk(ast.parse(code)):
312			if isinstance(node, ast.Name):
313			yield node.id
314
315			def _safe_eval(self, code, event):
316			"""
317			Try to evaluate the given code on the given frame. If failure occurs, returns some ugly string with exception.
318			"""
319			try:
320			return eval(code, event.globals if self.globals else {}, event.locals)
321			except Exception as exc:
322			return "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
323
324			def __call__(self, event):
325			"""
326			Handle event and print the specified variables.
327			"""
328			first = True
329			frame_symbols = set(event.locals)
330			if self.globals:
331			frame_symbols \|= set(event.globals)
332
333			for code, symbols in self.names.items():
334			try:
335			obj = eval(code, event.globals if self.globals else {}, event.locals)
336			except AttributeError:
337			continue
338			except Exception as exc:
339			printout = "{internal-failure}FAILED EVAL: {internal-detail}{!r}".format(exc, **self.event_colors)
340			else:
341			printout = self._safe_repr(obj)
342
343			if frame_symbols >= symbols:
344			self.stream.write("{:>{align}} {vars}{:9} {vars-name}{} {vars}=> {reset}{}{reset}\n".format(
345			"",
346			"vars" if first else "...",
347			code,
348			printout,
349			align=self.filename_alignment,
350			**self.event_colors
351			))
352			first = False
353

ionelmc / python-hunter

Push — master ( 70f629...0878c0 )

src.hunter.ColorStreamAction.stream() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like