src.hunter.CallPrinter.__call__() - Code Metrics - Inspection of "Add a "CallPrinter" (WIP)." - ionelmc/python-hunter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( b6996b...890ecc )

by Ionel Cristian

created 2016-01-22 01:52 UTC

src.hunter.CallPrinter.call() C

↳ Parent: src.hunter.CallPrinter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	40
rs	5.5
cc	7

from __future__ import absolute_import

import ast
import os
import pdb
import sys

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

DEFAULT_MIN_FILENAME_ALIGNMENT = 40
NO_COLORS = {
    'reset': '',
    'filename': '',
    'colon': '',
    'lineno': '',
    'kind': '',
    'continuation': '',
    'return': '',
    'exception': '',
    'detail': '',
    'vars': '',
    'vars-name': '',
    'call': '',
    'line': '',
    'internal-failure': '',
    'internal-detail': '',
    'source-failure': '',
    'source-detail': '',
}
EVENT_COLORS = {
    'reset': Style.RESET_ALL,
    'filename': '',
    'colon': Fore.BLACK + Style.BRIGHT,
    'lineno': Style.RESET_ALL,
    'kind': Fore.CYAN,
    'continuation': Fore.BLUE + Style.BRIGHT,
    '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': Back.RED + Style.BRIGHT + Fore.RED,
    'internal-detail': Fore.WHITE,
    'source-failure': Style.BRIGHT + Back.YELLOW + Fore.YELLOW,
    'source-detail': Fore.WHITE,
    'call': Fore.CYAN + Style.BRIGHT,
}
CODE_COLORS = {
    'call': Fore.RESET + Style.BRIGHT,
    'line': Fore.RESET,
    'return': Fore.YELLOW,
    'exception': Fore.RED,
}
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(Action):
    _stream_cache = {}
    _stream = None
    _tty = None
    default_stream = sys.stderr
    force_colors = False
    repr_limit = 1024

    @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)
            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 CallPrinter(Fields.stream.filename_alignment, 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``.
    """

    def __init__(self,
                 stream=ColorStreamAction.default_stream, force_colors=False,
                 filename_alignment=DEFAULT_MIN_FILENAME_ALIGNMENT):
        self.stream = stream
        self.force_colors = force_colors
        self.filename_alignment = max(5, filename_alignment)
        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 = event.filename or "<???>"
        if len(filename) > self.filename_alignment:
            filename = '[...]{}'.format(filename[5 - self.filename_alignment:])
        ident = event.module, event.function

        if event.kind == 'call':
            code = event.code
            self.stream.write("{filename}{0:>{align}}{colon}:{lineno}{1:<5} {kind}{2:9} {3}{call}=> {4}({vars}{5}{call}){reset}\n".format(
                filename,
                event.lineno,
                event.kind,
                '  ' * len(self.stack),
                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,
                code=self.code_colors[event.kind],
                **self.event_colors
            ))
            self.stack.append(ident)
        elif event.kind in ('return', 'exception'):
            if self.stack and self.stack[-1] == ident:
                self.stack.pop()
            self.stream.write("{filename}{0:>{align}}{colon}:{lineno}{1:<5} {kind}{2:9} {code}{3}{4} {5}: {reset}{6}\n".format(
                filename,
                event.lineno,
                event.kind,
                '  ' * len(self.stack),
                {'return': '<=', 'exception': '<<'}[event.kind],
                event.function,
                self._safe_repr(event.arg),
                align=self.filename_alignment,
                code=self.event_colors[event.kind],
                **self.event_colors
            ))


class CodePrinter(Fields.stream.filename_alignment, 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``.
    """

    def __init__(self,
                 stream=ColorStreamAction.default_stream, force_colors=False,
                 filename_alignment=DEFAULT_MIN_FILENAME_ALIGNMENT):
        self.stream = stream
        self.force_colors = force_colors
        self.filename_alignment = max(5, filename_alignment)

    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 __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 = event.filename or "<???>"
        if len(filename) > self.filename_alignment:
            filename = '[...]{}'.format(filename[5 - self.filename_alignment:])

        # 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(
            filename,
            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 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.
        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``.
        globals (bool): Allow access to globals. Default: ``False`` (only looks at locals).
    """

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

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

ionelmc / python-hunter

Push — master ( b6996b...890ecc )

src.hunter.CallPrinter.__call__() C

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like

src.hunter.CallPrinter.call() C