ColorStreamAction.stream() - Code Metrics - Inspection of "Update gitignore." - ionelmc/python-hunter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 449e73...c6f33b )

by Ionel Cristian

created 2016-04-13 21:20 UTC

ColorStreamAction.stream() C

↳ Parent: ColorStreamAction

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	7
dl	0
loc	3
rs	5.5

from __future__ import absolute_import

import ast
import os
import sys
import threading
from collections import defaultdict
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=lambda **kwargs: __import__('pdb').Pdb(**kwargs), **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 Manhole(Action):
    def __init__(self, **options):
        self.options = options

    def __call__(self, event):
        import manhole
        inst = manhole.install(strict=False, thread=False, **self.options)
        inst.handle_oneshot()


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

    def __init__(self,
                 stream=sys.stderr,
                 force_colors=False,
                 filename_alignment=40,
                 thread_alignment=12,
                 repr_limit=1024):
        self.force_colors = force_colors
        self.stream = stream
        self.filename_alignment = filename_alignment
        self.thread_alignment = thread_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)
        thread_name = threading.current_thread().name if event.tracer.threading_support else ''
        thread_align = self.thread_alignment if event.tracer.threading_support else ''

        self.stream.write(
            "{thread:{thread_align}}{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {code}{}{reset}\n".format(
                self._format_filename(event),
                event.lineno,
                event.kind,
                lines[0],
                thread=thread_name, thread_align=thread_align,
                align=self.filename_alignment,
                code=self.code_colors[event.kind],
                **self.event_colors
            ))
        for line in lines[1:]:
            self.stream.write("{thread:{thread_align}}{:>{align}}       {kind}{:9} {code}{}{reset}\n".format(
                "",
                r"   |",
                line,
                thread=thread_name, thread_align=thread_align,
                align=self.filename_alignment,
                code=self.code_colors[event.kind],
                **self.event_colors
            ))

        if event.kind in ('return', 'exception'):
            self.stream.write(
                "{thread:{thread_align}}{:>{align}}       {continuation}{:9} {color}{} "
                "value: {detail}{}{reset}\n".format(
                    "",
                    "...",
                    event.kind,
                    self._safe_repr(event.arg),
                    thread=thread_name, thread_align=thread_align,
                    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.locals = defaultdict(list)

    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
        thread = threading.current_thread()
        thread_name = thread.name if event.tracer.threading_support else ''
        thread_align = self.thread_alignment if event.tracer.threading_support else ''
        stack = self.locals[thread.ident]

        if event.kind == 'call':
            code = event.code
            stack.append(ident)
            self.stream.write(
                "{thread:{thread_align}}{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {}{call}=>{normal} "
                "{}({}{call}{normal}){reset}\n".format(
                    filename,
                    event.lineno,
                    event.kind,
                    '   ' * (len(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]),
                    thread=thread_name, thread_align=thread_align,
                    align=self.filename_alignment,
                    **self.event_colors
                ))
        elif event.kind in ('return', 'exception'):
            self.stream.write(
                "{thread:{thread_align}}{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} "
                "{code}{}{}{normal} {}: {reset}{}\n".format(
                    filename,
                    event.lineno,
                    event.kind,
                    '   ' * (len(stack) - 1),
                    {'return': '<=', 'exception': '<!'}[event.kind],
                    event.function,
                    self._safe_repr(event.arg),
                    thread=thread_name, thread_align=thread_align,
                    align=self.filename_alignment,
                    code=self.event_colors[event.kind],
                    **self.event_colors
                ))
            if stack and stack[-1] == ident:
                stack.pop()
        else:
            self.stream.write(
                "{thread:{thread_align}}{filename}{:>{align}}{colon}:{lineno}{:<5} {kind}{:9} {reset}{}{}\n".format(
                    filename,
                    event.lineno,
                    event.kind,
                    '   ' * len(stack),
                    event.source.strip(),
                    thread=thread_name, thread_align=thread_align,
                    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)
        thread_name = threading.current_thread().name if event.tracer.threading_support else ''
        thread_align = self.thread_alignment if event.tracer.threading_support else ''

        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("{thread:{thread_align}}{:>{align}}       {vars}{:9} {vars-name}{} {vars}=> {reset}{}{reset}\n".format(
                    "",
                    "vars" if first else "...",
                    code,
                    printout,
                    thread=thread_name, thread_align=thread_align,
                    align=self.filename_alignment,
                    **self.event_colors
                ))
                first = False


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

ionelmc / python-hunter

Push — master ( 449e73...c6f33b )

ColorStreamAction.stream() C

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like