src.hunter.Event.threadname() - Code Metrics - Inspection of "Add Event.thread/threadid/threadname (last two ava..." - ionelmc/python-hunter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( da4845...4a7506 )

by Ionel Cristian

created 2016-02-23 19:31 UTC

src.hunter.Event.threadname() A

↳ Parent: src.hunter.Event

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	6
rs	9.4285
cc	1

from __future__ import absolute_import

import linecache
import re
import tokenize
from functools import partial
from threading import current_thread
from threading import main_thread

from fields import Fields

from .const import SITE_PACKAGES_PATHS
from .const import SYS_PREFIX_PATHS
from .util import cached_property

STARTSWITH_TYPES = list, tuple, set


class Event(Fields.kind.function.module.filename):
    """
    Event wrapper for ``frame, kind, arg`` (the arguments the settrace function gets). This objects is passed to your
    custom functions or predicates.

    Provides few convenience properties.

    .. warning::

        Users do not instantiate this directly.
    """
    frame = None
    kind = None
    arg = None
    tracer = None
    threadid = None
    threadname = None

    def __init__(self, frame, kind, arg, tracer):
        self.frame = frame
        self.kind = kind
        self.arg = arg
        self.tracer = tracer

    @cached_property
    def threadid(self):
        """
        Current thread ident. If current thread is main thread then it returns ``None``.
        """
        main = main_thread().ident
        current = self.thread.ident
        return current if current != main else None

    @cached_property
    def threadname(self):
        """
        Current thread name.
        """
        return self.thread.name

    @cached_property
    def thread(self):
        """
        Current thread object.
        """
        return current_thread()

    @cached_property
    def locals(self):
        """
        A dict with local variables.
        """
        return self.frame.f_locals

    @cached_property
    def globals(self):
        """
        A dict with global variables.
        """
        return self.frame.f_globals

    @cached_property
    def function(self):
        """
        A string with function name.
        """
        return self.code.co_name

    @cached_property
    def module(self):
        """
        A string with module name (eg: ``"foo.bar"``).
        """
        module = self.frame.f_globals.get('__name__', '')
        if module is None:
            module = ''

        return module

    @cached_property
    def filename(self):
        """
        A string with absolute path to file.
        """
        filename = self.frame.f_globals.get('__file__', '')
        if filename is None:
            filename = ''

        if filename.endswith(('.pyc', '.pyo')):
            filename = filename[:-1]
        elif filename.endswith('$py.class'):  # Jython
            filename = filename[:-9] + ".py"

        return filename

    @cached_property
    def lineno(self):
        """
        An integer with line number in file.
        """
        return self.frame.f_lineno

    @cached_property
    def code(self):
        """
        A code object (not a string).
        """
        return self.frame.f_code

    @cached_property
    def stdlib(self):
        """
        A boolean flag. ``True`` if frame is in stdlib.
        """
        if self.filename.startswith(SITE_PACKAGES_PATHS):
            # if it's in site-packages then its definitely not stdlib
            return False
        elif self.filename.startswith(SYS_PREFIX_PATHS):
            return True
        else:
            return False

    @cached_property
    def fullsource(self):
        """
        A string with the sourcecode for the current statement (from ``linecache`` - failures are ignored).

        May include multiple lines if it's a class/function definition (will include decorators).
        """
        try:
            return self._raw_fullsource
        except Exception as exc:
            return "??? NO SOURCE: {!r}".format(exc)

    @cached_property
    def source(self, getline=linecache.getline):
        """
        A string with the sourcecode for the current line (from ``linecache`` - failures are ignored).

        Fast but sometimes incomplete.
        """
        try:
            return getline(self.filename, self.lineno)
        except Exception as exc:
            return "??? NO SOURCE: {!r}".format(exc)

    @cached_property
    def _raw_fullsource(self,
                        getlines=linecache.getlines,
                        getline=linecache.getline,
                        generate_tokens=tokenize.generate_tokens):
        if self.kind == 'call' and self.code.co_name != "<module>":
            lines = []
            try:
                for _, token, _, _, line in generate_tokens(partial(
                    next,
                    yield_lines(self.filename, self.lineno - 1, lines.append)
                )):
                    if token in ("def", "class", "lambda"):
                        return ''.join(lines)
            except tokenize.TokenError:
                pass

        return getline(self.filename, self.lineno)

    __getitem__ = object.__getattribute__


def yield_lines(filename, start, collector,
                limit=10,
                getlines=linecache.getlines,
                leading_whitespace_re=re.compile('(^[ \t]*)(?:[^ \t\n])', re.MULTILINE)):
    dedent = None
    amount = 0
    for line in getlines(filename)[start:start + limit]:
        if dedent is None:
            dedent = leading_whitespace_re.findall(line)
            dedent = dedent[0] if dedent else ""
            amount = len(dedent)
        elif not line.startswith(dedent):
            break
        collector(line)
        yield line[amount:]


1			from __future__ import absolute_import
2
3			import linecache
4			import re
5			import tokenize
6			from functools import partial
7			from threading import current_thread
8			from threading import main_thread
9
10			from fields import Fields
11
12			from .const import SITE_PACKAGES_PATHS
13			from .const import SYS_PREFIX_PATHS
14			from .util import cached_property
15
16			STARTSWITH_TYPES = list, tuple, set
17
18
19			class Event(Fields.kind.function.module.filename):
20			"""
21			Event wrapper for ``frame, kind, arg`` (the arguments the settrace function gets). This objects is passed to your
22			custom functions or predicates.
23
24			Provides few convenience properties.
25
26			.. warning::
27
28			Users do not instantiate this directly.
29			"""
30			frame = None
31			kind = None
32			arg = None
33			tracer = None
34			threadid = None
35			threadname = None
36
37			def __init__(self, frame, kind, arg, tracer):
38			self.frame = frame
39			self.kind = kind
40			self.arg = arg
41			self.tracer = tracer
42
43			@cached_property
44			def threadid(self):
45			"""
46			Current thread ident. If current thread is main thread then it returns ``None``.
47			"""
48			main = main_thread().ident
49			current = self.thread.ident
50			return current if current != main else None
51
52			@cached_property
53			def threadname(self):
54			"""
55			Current thread name.
56			"""
57			return self.thread.name
58
59			@cached_property
60			def thread(self):
61			"""
62			Current thread object.
63			"""
64			return current_thread()
65
66			@cached_property
67			def locals(self):
68			"""
69			A dict with local variables.
70			"""
71			return self.frame.f_locals
72
73			@cached_property
74			def globals(self):
75			"""
76			A dict with global variables.
77			"""
78			return self.frame.f_globals
79
80			@cached_property
81			def function(self):
82			"""
83			A string with function name.
84			"""
85			return self.code.co_name
86
87			@cached_property
88			def module(self):
89			"""
90			A string with module name (eg: ``"foo.bar"``).
91			"""
92			module = self.frame.f_globals.get('__name__', '')
93			if module is None:
94			module = ''
95
96			return module
97
98			@cached_property
99			def filename(self):
100			"""
101			A string with absolute path to file.
102			"""
103			filename = self.frame.f_globals.get('__file__', '')
104			if filename is None:
105			filename = ''
106
107			if filename.endswith(('.pyc', '.pyo')):
108			filename = filename[:-1]
109			elif filename.endswith('$py.class'): # Jython
110			filename = filename[:-9] + ".py"
111
112			return filename
113
114			@cached_property
115			def lineno(self):
116			"""
117			An integer with line number in file.
118			"""
119			return self.frame.f_lineno
120
121			@cached_property
122			def code(self):
123			"""
124			A code object (not a string).
125			"""
126			return self.frame.f_code
127
128			@cached_property
129			def stdlib(self):
130			"""
131			A boolean flag. ``True`` if frame is in stdlib.
132			"""
133			if self.filename.startswith(SITE_PACKAGES_PATHS):
134			# if it's in site-packages then its definitely not stdlib
135			return False
136			elif self.filename.startswith(SYS_PREFIX_PATHS):
137			return True
138			else:
139			return False
140
141			@cached_property
142			def fullsource(self):
143			"""
144			A string with the sourcecode for the current statement (from ``linecache`` - failures are ignored).
145
146			May include multiple lines if it's a class/function definition (will include decorators).
147			"""
148			try:
149			return self._raw_fullsource
150			except Exception as exc:
151			return "??? NO SOURCE: {!r}".format(exc)
152
153			@cached_property
154			def source(self, getline=linecache.getline):
155			"""
156			A string with the sourcecode for the current line (from ``linecache`` - failures are ignored).
157
158			Fast but sometimes incomplete.
159			"""
160			try:
161			return getline(self.filename, self.lineno)
162			except Exception as exc:
163			return "??? NO SOURCE: {!r}".format(exc)
164
165			@cached_property
166			def _raw_fullsource(self,
167			getlines=linecache.getlines,
168			getline=linecache.getline,
169			generate_tokens=tokenize.generate_tokens):
170			if self.kind == 'call' and self.code.co_name != "<module>":
171			lines = []
172			try:
173			for _, token, _, _, line in generate_tokens(partial(
174			next,
175			yield_lines(self.filename, self.lineno - 1, lines.append)
176			)):
177			if token in ("def", "class", "lambda"):
178			return ''.join(lines)
179			except tokenize.TokenError:
180			pass
181
182			return getline(self.filename, self.lineno)
183
184			__getitem__ = object.__getattribute__
185
186
187			def yield_lines(filename, start, collector,
188			limit=10,
189			getlines=linecache.getlines,
190			leading_whitespace_re=re.compile('(^[ \t]*)(?:[^ \t\n])', re.MULTILINE)):
191			dedent = None
192			amount = 0
193			for line in getlines(filename)[start:start + limit]:
194			if dedent is None:
195			dedent = leading_whitespace_re.findall(line)
196			dedent = dedent[0] if dedent else ""
197			amount = len(dedent)
198			elif not line.startswith(dedent):
199			break
200			collector(line)
201			yield line[amount:]
202

ionelmc / python-hunter

Push — master ( da4845...4a7506 )

src.hunter.Event.threadname() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like