Query.__init__() - Code Metrics - ionelmc/python-hunter - Measure and Improve Code Quality continuously with Scrutinizer

Query.init() F
last analyzed 2018-06-04 15:17 UTC

↳ Parent: Query

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	20
c	1
b	0
f	0
dl	0
loc	92
rs	2

How to fix Long Method Complexity

from __future__ import absolute_import

import inspect
import re
from itertools import chain

from fields import Fields
from six import string_types

from .actions import Action
from .event import Event

ALLOWED_KEYS = tuple(i for i in Event.__dict__.keys() if not i.startswith('_') and i not in ('tracer', 'thread'))
ALLOWED_OPERATORS = (
    'startswith', 'endswith', 'in', 'contains', 'regex',
    'sw', 'ew', 'has', 'rx',
    'gt', 'gte', 'lt', 'lte',
)


def _sloppy_hash(obj):
    try:
        return hash(obj)
    except TypeError:
        return 'id(%x)' % id(obj)


class Query(Fields.query_eq.query_startswith.query_endswith.query_in.query_contains):
    """
    A query class.

    See :class:`hunter.Event` for fields that can be filtered on.
    """
    def __init__(self, **query):
        """
        Args:
            query: criteria to match on.

                Accepted arguments:
                ``arg``,
                ``calls``,
                ``code``,
                ``depth``,
                ``filename``,
                ``frame``,
                ``fullsource``,
                ``function``,
                ``globals``,
                ``kind``,
                ``lineno``,
                ``locals``,
                ``module``,
                ``source``,
                ``stdlib``,
                ``threadid``,
                ``threadname``.
        """
        query_eq = {}
        query_startswith = {}
        query_endswith = {}
        query_in = {}
        query_contains = {}
        query_regex = {}
        query_lt = {}
        query_lte = {}
        query_gt = {}
        query_gte = {}

        for key, value in query.items():
            parts = [p for p in key.split('_') if p]
            count = len(parts)
            if count > 2:
                raise TypeError('Unexpected argument %r. Must be one of %s with optional operators like: %s' % (
                    key, ALLOWED_KEYS, ALLOWED_OPERATORS
                ))
            elif count == 2:
                prefix, operator = parts
                if operator in ('startswith', 'sw'):
                    if not isinstance(value, string_types):
                        if not isinstance(value, (list, set, tuple)):
                            raise ValueError('Value %r for %r is invalid. Must be a string, list, tuple or set.' % (value, key))
                        value = tuple(value)
                    mapping = query_startswith
                elif operator in ('endswith', 'ew'):
                    if not isinstance(value, string_types):
                        if not isinstance(value, (list, set, tuple)):
                            raise ValueError('Value %r for %r is invalid. Must be a string, list, tuple or set.' % (value, key))
                        value = tuple(value)
                    mapping = query_endswith
                elif operator == 'in':
                    mapping = query_in
                elif operator in ('contains', 'has'):
                    mapping = query_contains
                elif operator in ('regex', 'rx'):
                    value = re.compile(value)
                    mapping = query_regex
                elif operator == 'lt':
                    mapping = query_lt
                elif operator == 'lte':
                    mapping = query_lte
                elif operator == 'gt':
                    mapping = query_gt
                elif operator == 'gte':
                    mapping = query_gte
                else:
                    raise TypeError('Unexpected operator %r. Must be one of %s.' % (operator, ALLOWED_OPERATORS))
            else:
                mapping = query_eq
                prefix = key

            if prefix not in ALLOWED_KEYS:
                raise TypeError('Unexpected argument %r. Must be one of %s.' % (key, ALLOWED_KEYS))

            mapping[prefix] = value

        self.query_eq = tuple(sorted(query_eq.items()))
        self.query_startswith = tuple(sorted(query_startswith.items()))
        self.query_endswith = tuple(sorted(query_endswith.items()))
        self.query_in = tuple(sorted(query_in.items()))
        self.query_contains = tuple(sorted(query_contains.items()))
        self.query_regex = tuple(sorted(query_regex.items()))
        self.query_lt = tuple(sorted(query_lt.items()))
        self.query_lte = tuple(sorted(query_lte.items()))
        self.query_gt = tuple(sorted(query_gt.items()))
        self.query_gte = tuple(sorted(query_gte.items()))

    def __str__(self):

        return 'Query(%s)' % (
            ', '.join(
                ', '.join('%s%s=%r' % (key, kind, value) for key, value in mapping)
                for kind, mapping in [
                    ('', self.query_eq),
                    ('_in', self.query_in),
                    ('_contains', self.query_contains),
                    ('_startswith', self.query_startswith),
                    ('_endswith', self.query_endswith),
                    ('_regex', self.query_regex),
                    ('_lt', self.query_lt),
                    ('_lte', self.query_lte),
                    ('_gt', self.query_gt),
                    ('_gte', self.query_gte),
                ] if mapping
            )
        )

    def __repr__(self):

        return '<hunter.predicates.Query: %s>' % ' '.join(
            fmt % (mapping,) for fmt, mapping in [
                ('query_eq=%r', self.query_eq),
                ('query_in=%r', self.query_in),
                ('query_contains=%r', self.query_contains),
                ('query_startswith=%r', self.query_startswith),
                ('query_endswith=%r', self.query_endswith),
                ('query_regex=%r', self.query_regex),
                ('query_regex=%r', self.query_regex),
                ('query_lt=%r', self.query_lt),
                ('query_lte=%r', self.query_lte),
                ('query_gt=%r', self.query_gt),
                ('query_gte=%r', self.query_gte),
            ] if mapping
        )

    def __call__(self, event):
        """
        Handles event. Returns True if all criteria matched.
        """
        for key, value in self.query_eq:
            evalue = event[key]
            if evalue != value:
                return False
        for key, value in self.query_in:
            evalue = event[key]
            if evalue not in value:
                return False
        for key, value in self.query_contains:
            evalue = event[key]
            if value not in evalue:
                return False
        for key, value in self.query_startswith:
            evalue = event[key]
            if not evalue.startswith(value):
                return False
        for key, value in self.query_endswith:
            evalue = event[key]
            if not evalue.endswith(value):
                return False
        for key, value in self.query_regex:
            evalue = event[key]
            if not value.match(evalue):
                return False
        for key, value in self.query_gt:
            evalue = event[key]
            if not evalue > value:
                return False
        for key, value in self.query_gte:
            evalue = event[key]
            if not evalue >= value:
                return False
        for key, value in self.query_lt:
            evalue = event[key]
            if not evalue < value:
                return False
        for key, value in self.query_lte:
            evalue = event[key]
            if not evalue <= value:
                return False

        return True

    def __or__(self, other):
        """
        Convenience API so you can do ``Q() | Q()``. It converts that to ``Or(Q(), Q())``.
        """
        return Or(self, other)

    def __and__(self, other):
        """
        Convenience API so you can do ``Q() & Q()``. It converts that to ``And(Q(), Q())``.
        """
        return And(self, other)

    def __invert__(self):
        return Not(self)

    __ror__ = __or__
    __rand__ = __and__


class When(Fields.condition.actions):
    """
    Runs ``actions`` when ``condition(event)`` is ``True``.

    Actions take a single ``event`` argument.
    """

    def __init__(self, condition, *actions):
        if not actions:
            raise TypeError('Must give at least one action.')
        super(When, self).__init__(condition, tuple(
            action() if inspect.isclass(action) and issubclass(action, Action) else action
            for action in actions))

    def __str__(self):
        return 'When(%s, %s)' % (
            self.condition,
            ', '.join(repr(p) for p in self.actions)
        )

    def __repr__(self):
        return '<hunter.predicates.When: condition=%r, actions=%r>' % (self.condition, self.actions)

    def __call__(self, event):
        """
        Handles the event.
        """
        if self.condition(event):
            for action in self.actions:
                action(event)
            return True
        else:
            return False

    def __or__(self, other):
        return Or(self, other)

    def __and__(self, other):
        return And(self, other)

    def __invert__(self):
        return Not(self)

    __ror__ = __or__
    __rand__ = __and__


class And(Fields.predicates):

    """
    `And` predicate. Exits at the first sub-predicate that returns ``False``.
    """

    def __init__(self, *predicates):
        self.predicates = predicates

    def __str__(self):
        return 'And(%s)' % ', '.join(str(p) for p in self.predicates)

    def __repr__(self):
        return '<hunter.predicates.And: predicates=%r>' % (self.predicates,)

    def __call__(self, event):
        """
        Handles the event.
        """
        for predicate in self.predicates:
            if not predicate(event):
                return False
        else:
            return True

    def __eq__(self, other):
        if isinstance(other, And):
            if len(self.predicates) != len(other.predicates):
                return False
            return set(self.predicates) == set(other.predicates)
        return NotImplemented

    def __or__(self, other):
        return Or(self, other)

    def __and__(self, other):
        return And(*chain(self.predicates, other.predicates if isinstance(other, And) else (other,)))

    def __invert__(self):
        return Not(self)

    def __hash__(self):
        return hash(frozenset(self.predicates))

    __ror__ = __or__
    __rand__ = __and__


class Or(Fields.predicates):

    """
    `Or` predicate. Exits at first sub-predicate that returns ``True``.
    """

    def __init__(self, *predicates):
        self.predicates = predicates

    def __str__(self):
        return 'Or(%s)' % ', '.join(str(p) for p in self.predicates)

    def __repr__(self):
        return '<hunter.predicates.Or: predicates=%r>' % (self.predicates,)

    def __call__(self, event):
        """
        Handles the event.
        """
        for predicate in self.predicates:
            if predicate(event):
                return True
        else:
            return False

    def __eq__(self, other):
        if isinstance(other, Or):
            if len(self.predicates) != len(other.predicates):
                return False
            return set(self.predicates) == set(other.predicates)
        return NotImplemented

    def __or__(self, other):
        return Or(*chain(self.predicates, other.predicates if isinstance(other, Or) else (other,)))

    def __and__(self, other):
        return And(self, other)

    def __invert__(self):
        return Not(self)

    def __hash__(self):
        return hash(frozenset(self.predicates))

    __ror__ = __or__
    __rand__ = __and__


class Not(Fields.predicate):
    """
    `Not` predicate.
    """

    def __str__(self):
        return 'Not(%s)' % self.predicate

    def __repr__(self):
        return '<hunter.predicates.Not: predicate=%r>' % self.predicate

    def __call__(self, event):
        """
        Handles the event.
        """
        return not self.predicate(event)

    def __or__(self, other):
        if isinstance(other, Not):
            return Not(And(self.predicate, other.predicate))
        else:
            return Or(self, other)

    def __and__(self, other):
        if isinstance(other, Not):
            return Not(Or(self.predicate, other.predicate))
        else:
            return And(self, other)

    def __invert__(self):
        return self.predicate

    __ror__ = __or__
    __rand__ = __and__


1		from __future__ import absolute_import
2
3		import inspect
4		import re
5		from itertools import chain
6
7		from fields import Fields
8		from six import string_types
9
10		from .actions import Action
11		from .event import Event
12
13		ALLOWED_KEYS = tuple(i for i in Event.__dict__.keys() if not i.startswith('_') and i not in ('tracer', 'thread'))
14		ALLOWED_OPERATORS = (
15		'startswith', 'endswith', 'in', 'contains', 'regex',
16		'sw', 'ew', 'has', 'rx',
17		'gt', 'gte', 'lt', 'lte',
18		)
19
20
21		def _sloppy_hash(obj):
22		try:
23		return hash(obj)
24		except TypeError:
25		return 'id(%x)' % id(obj)
26
27
28		class Query(Fields.query_eq.query_startswith.query_endswith.query_in.query_contains):
29		"""
30		A query class.
31
32		See :class:`hunter.Event` for fields that can be filtered on.
33		"""
34		def __init__(self, **query):
35		"""
36		Args:
37		query: criteria to match on.
38
39		Accepted arguments:
40		``arg``,
41		``calls``,
42		``code``,
43		``depth``,
44		``filename``,
45		``frame``,
46		``fullsource``,
47		``function``,
48		``globals``,
49		``kind``,
50		``lineno``,
51		``locals``,
52		``module``,
53		``source``,
54		``stdlib``,
55		``threadid``,
56		``threadname``.
57		"""
58		query_eq = {}
59		query_startswith = {}
60		query_endswith = {}
61		query_in = {}
62		query_contains = {}
63		query_regex = {}
64		query_lt = {}
65		query_lte = {}
66		query_gt = {}
67		query_gte = {}
68
69		for key, value in query.items():
70		parts = [p for p in key.split('_') if p]
71		count = len(parts)
72		if count > 2:
73		raise TypeError('Unexpected argument %r. Must be one of %s with optional operators like: %s' % (
74		key, ALLOWED_KEYS, ALLOWED_OPERATORS
75		))
76		elif count == 2:
77		prefix, operator = parts
78		if operator in ('startswith', 'sw'):
79		if not isinstance(value, string_types):
80		if not isinstance(value, (list, set, tuple)):
81		raise ValueError('Value %r for %r is invalid. Must be a string, list, tuple or set.' % (value, key))
82		value = tuple(value)
83		mapping = query_startswith
84		elif operator in ('endswith', 'ew'):
85		if not isinstance(value, string_types):
86		if not isinstance(value, (list, set, tuple)):
87		raise ValueError('Value %r for %r is invalid. Must be a string, list, tuple or set.' % (value, key))
88		value = tuple(value)
89		mapping = query_endswith
90		elif operator == 'in':
91		mapping = query_in
92		elif operator in ('contains', 'has'):
93		mapping = query_contains
94		elif operator in ('regex', 'rx'):
95		value = re.compile(value)
96		mapping = query_regex
97		elif operator == 'lt':
98		mapping = query_lt
99		elif operator == 'lte':
100		mapping = query_lte
101		elif operator == 'gt':
102		mapping = query_gt
103		elif operator == 'gte':
104		mapping = query_gte
105		else:
106		raise TypeError('Unexpected operator %r. Must be one of %s.' % (operator, ALLOWED_OPERATORS))
107		else:
108		mapping = query_eq
109		prefix = key
110
111		if prefix not in ALLOWED_KEYS:
112		raise TypeError('Unexpected argument %r. Must be one of %s.' % (key, ALLOWED_KEYS))
113
114		mapping[prefix] = value
115
116		self.query_eq = tuple(sorted(query_eq.items()))
117		self.query_startswith = tuple(sorted(query_startswith.items()))
118		self.query_endswith = tuple(sorted(query_endswith.items()))
119		self.query_in = tuple(sorted(query_in.items()))
120		self.query_contains = tuple(sorted(query_contains.items()))
121		self.query_regex = tuple(sorted(query_regex.items()))
122		self.query_lt = tuple(sorted(query_lt.items()))
123		self.query_lte = tuple(sorted(query_lte.items()))
124		self.query_gt = tuple(sorted(query_gt.items()))
125		self.query_gte = tuple(sorted(query_gte.items()))
126
127	View Code Duplication	def __str__(self):
		0 ignored issues – show Duplication introduced 2017-09-01 15:14 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
128		return 'Query(%s)' % (
129		', '.join(
130		', '.join('%s%s=%r' % (key, kind, value) for key, value in mapping)
131		for kind, mapping in [
132		('', self.query_eq),
133		('_in', self.query_in),
134		('_contains', self.query_contains),
135		('_startswith', self.query_startswith),
136		('_endswith', self.query_endswith),
137		('_regex', self.query_regex),
138		('_lt', self.query_lt),
139		('_lte', self.query_lte),
140		('_gt', self.query_gt),
141		('_gte', self.query_gte),
142		] if mapping
143		)
144		)
145
146	View Code Duplication	def __repr__(self):
		0 ignored issues – show Duplication introduced 2017-09-01 15:14 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
147		return '<hunter.predicates.Query: %s>' % ' '.join(
148		fmt % (mapping,) for fmt, mapping in [
149		('query_eq=%r', self.query_eq),
150		('query_in=%r', self.query_in),
151		('query_contains=%r', self.query_contains),
152		('query_startswith=%r', self.query_startswith),
153		('query_endswith=%r', self.query_endswith),
154		('query_regex=%r', self.query_regex),
155		('query_regex=%r', self.query_regex),
156		('query_lt=%r', self.query_lt),
157		('query_lte=%r', self.query_lte),
158		('query_gt=%r', self.query_gt),
159		('query_gte=%r', self.query_gte),
160		] if mapping
161		)
162
163		def __call__(self, event):
164		"""
165		Handles event. Returns True if all criteria matched.
166		"""
167		for key, value in self.query_eq:
168		evalue = event[key]
169		if evalue != value:
170		return False
171		for key, value in self.query_in:
172		evalue = event[key]
173		if evalue not in value:
174		return False
175		for key, value in self.query_contains:
176		evalue = event[key]
177		if value not in evalue:
178		return False
179		for key, value in self.query_startswith:
180		evalue = event[key]
181		if not evalue.startswith(value):
182		return False
183		for key, value in self.query_endswith:
184		evalue = event[key]
185		if not evalue.endswith(value):
186		return False
187		for key, value in self.query_regex:
188		evalue = event[key]
189		if not value.match(evalue):
190		return False
191		for key, value in self.query_gt:
192		evalue = event[key]
193		if not evalue > value:
194		return False
195		for key, value in self.query_gte:
196		evalue = event[key]
197		if not evalue >= value:
198		return False
199		for key, value in self.query_lt:
200		evalue = event[key]
201		if not evalue < value:
202		return False
203		for key, value in self.query_lte:
204		evalue = event[key]
205		if not evalue <= value:
206		return False
207
208		return True
209
210		def __or__(self, other):
211		"""
212		Convenience API so you can do ``Q() \| Q()``. It converts that to ``Or(Q(), Q())``.
213		"""
214		return Or(self, other)
215
216		def __and__(self, other):
217		"""
218		Convenience API so you can do ``Q() & Q()``. It converts that to ``And(Q(), Q())``.
219		"""
220		return And(self, other)
221
222		def __invert__(self):
223		return Not(self)
224
225		__ror__ = __or__
226		__rand__ = __and__
227
228
229		class When(Fields.condition.actions):
230		"""
231		Runs ``actions`` when ``condition(event)`` is ``True``.
232
233		Actions take a single ``event`` argument.
234		"""
235
236		def __init__(self, condition, *actions):
237		if not actions:
238		raise TypeError('Must give at least one action.')
239		super(When, self).__init__(condition, tuple(
240		action() if inspect.isclass(action) and issubclass(action, Action) else action
241		for action in actions))
242
243		def __str__(self):
244		return 'When(%s, %s)' % (
245		self.condition,
246		', '.join(repr(p) for p in self.actions)
247		)
248
249		def __repr__(self):
250		return '<hunter.predicates.When: condition=%r, actions=%r>' % (self.condition, self.actions)
251
252		def __call__(self, event):
253		"""
254		Handles the event.
255		"""
256		if self.condition(event):
257		for action in self.actions:
258		action(event)
259		return True
260		else:
261		return False
262
263		def __or__(self, other):
264		return Or(self, other)
265
266		def __and__(self, other):
267		return And(self, other)
268
269		def __invert__(self):
270		return Not(self)
271
272		__ror__ = __or__
273		__rand__ = __and__
274
275
276	View Code Duplication	class And(Fields.predicates):
		0 ignored issues – show Duplication introduced 2016-08-10 22:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
277		"""
278		`And` predicate. Exits at the first sub-predicate that returns ``False``.
279		"""
280
281		def __init__(self, *predicates):
282		self.predicates = predicates
283
284		def __str__(self):
285		return 'And(%s)' % ', '.join(str(p) for p in self.predicates)
286
287		def __repr__(self):
288		return '<hunter.predicates.And: predicates=%r>' % (self.predicates,)
289
290		def __call__(self, event):
291		"""
292		Handles the event.
293		"""
294		for predicate in self.predicates:
295		if not predicate(event):
296		return False
297		else:
298		return True
299
300		def __eq__(self, other):
301		if isinstance(other, And):
302		if len(self.predicates) != len(other.predicates):
303		return False
304		return set(self.predicates) == set(other.predicates)
305		return NotImplemented
306
307		def __or__(self, other):
308		return Or(self, other)
309
310		def __and__(self, other):
311		return And(*chain(self.predicates, other.predicates if isinstance(other, And) else (other,)))
312
313		def __invert__(self):
314		return Not(self)
315
316		def __hash__(self):
317		return hash(frozenset(self.predicates))
318
319		__ror__ = __or__
320		__rand__ = __and__
321
322
323	View Code Duplication	class Or(Fields.predicates):
		0 ignored issues – show Duplication introduced 2016-08-10 22:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
324		"""
325		`Or` predicate. Exits at first sub-predicate that returns ``True``.
326		"""
327
328		def __init__(self, *predicates):
329		self.predicates = predicates
330
331		def __str__(self):
332		return 'Or(%s)' % ', '.join(str(p) for p in self.predicates)
333
334		def __repr__(self):
335		return '<hunter.predicates.Or: predicates=%r>' % (self.predicates,)
336
337		def __call__(self, event):
338		"""
339		Handles the event.
340		"""
341		for predicate in self.predicates:
342		if predicate(event):
343		return True
344		else:
345		return False
346
347		def __eq__(self, other):
348		if isinstance(other, Or):
349		if len(self.predicates) != len(other.predicates):
350		return False
351		return set(self.predicates) == set(other.predicates)
352		return NotImplemented
353
354		def __or__(self, other):
355		return Or(*chain(self.predicates, other.predicates if isinstance(other, Or) else (other,)))
356
357		def __and__(self, other):
358		return And(self, other)
359
360		def __invert__(self):
361		return Not(self)
362
363		def __hash__(self):
364		return hash(frozenset(self.predicates))
365
366		__ror__ = __or__
367		__rand__ = __and__
368
369
370		class Not(Fields.predicate):
371		"""
372		`Not` predicate.
373		"""
374
375		def __str__(self):
376		return 'Not(%s)' % self.predicate
377
378		def __repr__(self):
379		return '<hunter.predicates.Not: predicate=%r>' % self.predicate
380
381		def __call__(self, event):
382		"""
383		Handles the event.
384		"""
385		return not self.predicate(event)
386
387		def __or__(self, other):
388		if isinstance(other, Not):
389		return Not(And(self.predicate, other.predicate))
390		else:
391		return Or(self, other)
392
393		def __and__(self, other):
394		if isinstance(other, Not):
395		return Not(Or(self.predicate, other.predicate))
396		else:
397		return And(self, other)
398
399		def __invert__(self):
400		return self.predicate
401
402		__ror__ = __or__
403		__rand__ = __and__
404

ionelmc / python-hunter

Query.__init__() F last analyzed 2018-06-04 15:17 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

Query.init() F
last analyzed 2018-06-04 15:17 UTC