src.hunter.Query.__call__() - Code Metrics - Inspection of "Tweak a bit the new perf test." - ionelmc/python-hunter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 67ac21...36843e )

by Ionel Cristian

created 2015-12-23 01:54 UTC

src.hunter.Query.call() F

↳ Parent: src.hunter.Query

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	30
rs	2.7716
cc	13

How to fix Complexity

from __future__ import absolute_import

import inspect
import re

from fields import Fields
from itertools import chain
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('_'))
ALLOWED_OPERATORS = 'startswith', 'endswith', 'in', 'contains', 'regex'


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``, ``code``, ``filename``, ``frame``, ``fullsource``, ``function``,
                ``globals``, ``kind``, ``lineno``, ``locals``, ``module``, ``source``, ``stdlib``, ``tracer``.
        """
        self.query_eq = {}
        self.query_startswith = {}
        self.query_endswith = {}
        self.query_in = {}
        self.query_contains = {}
        self.query_regex = {}

        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 not in ALLOWED_OPERATORS:
                    raise TypeError('Unexpected operator %r. Must be one of %s.'.format(operator, ALLOWED_OPERATORS))
                elif operator == 'startswith':
                    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 = self.query_startswith
                elif operator == 'endswith':
                    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 = self.query_endswith
                elif operator == 'in':
                    mapping = self.query_in
                elif operator == 'contains':
                    mapping = self.query_contains
                elif operator == 'regex':
                    value = re.compile(value)
                    mapping = self.query_regex
            else:
                mapping = self.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

    def __str__(self):
        return 'Query(%s)' % (
            ', '.join(
                ', '.join('%s%s=%r' % (key, kind, value) for key, value in mapping.items())
                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),
                ] 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),
            ] if mapping
        )

    def __call__(self, event):
        """
        Handles event. Returns True if all criteria matched.
        """
        for key, value in self.query_eq.items():
            evalue = event[key]
            if evalue != value:
                return False
        for key, value in self.query_in.items():
            evalue = event[key]
            if evalue not in value:
                return False
        for key, value in self.query_contains.items():
            evalue = event[key]
            if value not in evalue:
                return False
        for key, value in self.query_startswith.items():
            evalue = event[key]
            if not evalue.startswith(value):
                return False
        for key, value in self.query_endswith.items():
            evalue = event[key]
            if not evalue.endswith(value):
                return False
        for key, value in self.query_regex.items():
            evalue = event[key]
            if not value.match(evalue):
                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)


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, [
            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)


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 __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)


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 __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)


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


1			from __future__ import absolute_import
2
3			import inspect
4			import re
5
6			from fields import Fields
7			from itertools import chain
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('_'))
14			ALLOWED_OPERATORS = 'startswith', 'endswith', 'in', 'contains', 'regex'
15
16
17			class Query(Fields.query_eq.query_startswith.query_endswith.query_in.query_contains):
18			"""
19			A query class.
20
21			See :class:`hunter.Event` for fields that can be filtered on.
22			"""
23			def __init__(self, **query):
24			"""
25			Args:
26			query: criteria to match on.
27
28			Accepted arguments: ``arg``, ``code``, ``filename``, ``frame``, ``fullsource``, ``function``,
29			``globals``, ``kind``, ``lineno``, ``locals``, ``module``, ``source``, ``stdlib``, ``tracer``.
30			"""
31			self.query_eq = {}
32			self.query_startswith = {}
33			self.query_endswith = {}
34			self.query_in = {}
35			self.query_contains = {}
36			self.query_regex = {}
37
38			for key, value in query.items():
39			parts = [p for p in key.split('_') if p]
40			count = len(parts)
41			if count > 2:
42			raise TypeError('Unexpected argument %r. Must be one of %s with optional operators like: %s' % (
43			key, ALLOWED_KEYS, ALLOWED_OPERATORS
44			))
45			elif count == 2:
46			prefix, operator = parts
47			if operator not in ALLOWED_OPERATORS:
48			raise TypeError('Unexpected operator %r. Must be one of %s.'.format(operator, ALLOWED_OPERATORS))
49			elif operator == 'startswith':
50			if not isinstance(value, string_types):
51			if not isinstance(value, (list, set, tuple)):
52			raise ValueError('Value %r for %r is invalid. Must be a string, list, tuple or set.' % (value, key))
53			value = tuple(value)
54			mapping = self.query_startswith
55			elif operator == 'endswith':
56			if not isinstance(value, string_types):
57			if not isinstance(value, (list, set, tuple)):
58			raise ValueError('Value %r for %r is invalid. Must be a string, list, tuple or set.' % (value, key))
59			value = tuple(value)
60			mapping = self.query_endswith
61			elif operator == 'in':
62			mapping = self.query_in
63			elif operator == 'contains':
64			mapping = self.query_contains
65			elif operator == 'regex':
66			value = re.compile(value)
67			mapping = self.query_regex
68			else:
69			mapping = self.query_eq
70			prefix = key
71
72			if prefix not in ALLOWED_KEYS:
73			raise TypeError('Unexpected argument %r. Must be one of %s.' % (key, ALLOWED_KEYS))
74
75			mapping[prefix] = value
76
77			def __str__(self):
78			return 'Query(%s)' % (
79			', '.join(
80			', '.join('%s%s=%r' % (key, kind, value) for key, value in mapping.items())
81			for kind, mapping in [
82			('', self.query_eq),
83			('_in', self.query_in),
84			('_contains', self.query_contains),
85			('_startswith', self.query_startswith),
86			('_endswith', self.query_endswith),
87			('_regex', self.query_regex),
88			] if mapping
89			)
90			)
91
92			def __repr__(self):
93			return '<hunter._predicates.Query: %s>' % ' '.join(
94			fmt % mapping for fmt, mapping in [
95			('query_eq=%r', self.query_eq),
96			('query_in=%r', self.query_in),
97			('query_contains=%r', self.query_contains),
98			('query_startswith=%r', self.query_startswith),
99			('query_endswith=%r', self.query_endswith),
100			('query_regex=%r', self.query_regex),
101			] if mapping
102			)
103
104			def __call__(self, event):
105			"""
106			Handles event. Returns True if all criteria matched.
107			"""
108			for key, value in self.query_eq.items():
109			evalue = event[key]
110			if evalue != value:
111			return False
112			for key, value in self.query_in.items():
113			evalue = event[key]
114			if evalue not in value:
115			return False
116			for key, value in self.query_contains.items():
117			evalue = event[key]
118			if value not in evalue:
119			return False
120			for key, value in self.query_startswith.items():
121			evalue = event[key]
122			if not evalue.startswith(value):
123			return False
124			for key, value in self.query_endswith.items():
125			evalue = event[key]
126			if not evalue.endswith(value):
127			return False
128			for key, value in self.query_regex.items():
129			evalue = event[key]
130			if not value.match(evalue):
131			return False
132
133			return True
134
135			def __or__(self, other):
136			"""
137			Convenience API so you can do ``Q() \| Q()``. It converts that to ``Or(Q(), Q())``.
138			"""
139			return Or(self, other)
140
141			def __and__(self, other):
142			"""
143			Convenience API so you can do ``Q() & Q()``. It converts that to ``And(Q(), Q())``.
144			"""
145			return And(self, other)
146
147			def __invert__(self):
148			return Not(self)
149
150
151			class When(Fields.condition.actions):
152			"""
153			Runs ``actions`` when ``condition(event)`` is ``True``.
154
155			Actions take a single ``event`` argument.
156			"""
157
158			def __init__(self, condition, *actions):
159			if not actions:
160			raise TypeError('Must give at least one action.')
161			super(When, self).__init__(condition, [
162			action() if inspect.isclass(action) and issubclass(action, Action) else action
163			for action in actions
164			])
165
166			def __str__(self):
167			return 'When(%s, %s)' % (
168			self.condition,
169			', '.join(repr(p) for p in self.actions)
170			)
171
172			def __repr__(self):
173			return '<hunter._predicates.When: condition=%r, actions=%r>' % (self.condition, self.actions)
174
175			def __call__(self, event):
176			"""
177			Handles the event.
178			"""
179			if self.condition(event):
180			for action in self.actions:
181			action(event)
182			return True
183			else:
184			return False
185
186			def __or__(self, other):
187			return Or(self, other)
188
189			def __and__(self, other):
190			return And(self, other)
191
192
193			class And(Fields.predicates):
194			"""
195			`And` predicate. Exits at the first sub-predicate that returns ``False``.
196			"""
197
198			def __init__(self, *predicates):
199			self.predicates = predicates
200
201			def __str__(self):
202			return 'And(%s)' % ', '.join(str(p) for p in self.predicates)
203
204			def __repr__(self):
205			return '<hunter._predicates.And: predicates=%r>' % (self.predicates,)
206
207			def __call__(self, event):
208			"""
209			Handles the event.
210			"""
211			for predicate in self.predicates:
212			if not predicate(event):
213			return False
214			else:
215			return True
216
217			def __or__(self, other):
218			return Or(self, other)
219
220			def __and__(self, other):
221			return And(*chain(self.predicates, other.predicates if isinstance(other, And) else (other,)))
222
223			def __invert__(self):
224			return Not(self)
225
226
227			class Or(Fields.predicates):
228			"""
229			`Or` predicate. Exits at first sub-predicate that returns ``True``.
230			"""
231
232			def __init__(self, *predicates):
233			self.predicates = predicates
234
235			def __str__(self):
236			return 'Or(%s)' % ', '.join(str(p) for p in self.predicates)
237
238			def __repr__(self):
239			return '<hunter._predicates.Or: predicates=%r>' % (self.predicates,)
240
241			def __call__(self, event):
242			"""
243			Handles the event.
244			"""
245			for predicate in self.predicates:
246			if predicate(event):
247			return True
248			else:
249			return False
250
251			def __or__(self, other):
252			return Or(*chain(self.predicates, other.predicates if isinstance(other, Or) else (other,)))
253
254			def __and__(self, other):
255			return And(self, other)
256
257			def __invert__(self):
258			return Not(self)
259
260
261			class Not(Fields.predicate):
262			"""
263			`Not` predicate.
264			"""
265
266			def __str__(self):
267			return 'Not(%s)' % self.predicate
268
269			def __repr__(self):
270			return '<hunter._predicates.Not: predicate=%r>' % self.predicate
271
272			def __call__(self, event):
273			"""
274			Handles the event.
275			"""
276			return not self.predicate(event)
277
278			def __or__(self, other):
279			if isinstance(other, Not):
280			return Not(And(self.predicate, other.predicate))
281			else:
282			return Or(self, other)
283
284			def __and__(self, other):
285			if isinstance(other, Not):
286			return Not(Or(self.predicate, other.predicate))
287			else:
288			return And(self, other)
289
290			def __invert__(self):
291			return self.predicate
292

ionelmc / python-hunter

Push — master ( 67ac21...36843e )

src.hunter.Query.__call__() F

Complexity

Size

Duplication

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like

src.hunter.Query.call() F