3683952713

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/kobotoolbox/kpi/commit/<a class=hub.com/kobotoolbox/kpi/commit/045970b970d85fa5fe8f3a187d3974b152795a95">045970b97<a href="https://github.com/kobotoolbox/kpi/commit/045970b970d85fa5fe8f3a187d3974b152795a95">">Merge </a><a class="double-link" href="https://github.com/kobotoolbox/kpi/commit/<a class="double-link" href="https://github.com/kobotoolbox/kpi/commit/5fdd2411e309a30ab237083fc71721444acf8f0a">5fdd2411e</a>">5fdd2411e</a><a href="https://github.com/kobotoolbox/kpi/commit/045970b970d85fa5fe8f3a187d3974b152795a95"> into b30a88f5a">b30a88f5a</a>

Pull Request Pull Request #4182: Custom Views: integrate endpoints into front-end

Run Details

2553 of 4310 branches covered (59.23%)

537 of 537 new or added lines in 28 files covered. (100.0%)

11189 of 14824 relevant lines covered (75.48%)

1.51 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.5

/kpi/utils/query_parser/query_parser.py

# coding: utf-8
import operator
import re
from collections import defaultdict
from distutils import util
from functools import reduce

from django.conf import settings
from django.db.models import Q
from django_request_cache import cache_for_request

from kpi.exceptions import (
    SearchQueryTooShortException,
    QueryParserBadSyntax,
    QueryParserNotSupportedFieldLookup,
)
from .canopy_autogenerated_parser import parse as grammar_parse


"""
This is a utility for parsing a Boolean, Whoosh-like query string and
translating it into a Django Q object, which can then be used to filter a
queryset in the ORM.

Syntax examples:
    * `name:term` returns any object whose `name` field exactly matches
        `term` (case sensitive)
    * `owner__username=meg` traverses the `owner` relationship, returning
        any object whose `owner` field matches an object whose `username` field
        exactly matches `meg` (case sensitive)
    * `color:orange NOT (type:fruit OR type:chair)` returns anything
        whose color is orange so long as it is not a fruit or a chair. AND, OR,
        and NOT operators are supported. They must be in ALL CAPS.

Special notes:
    * The value `null` in a query is translated to `None`, e.g. `parent:null`
        effectively becomes the ORM `filter(parent=None)`
"""


class QueryParseActions:
    """
    Actions for the parser to take when it encounters certain identifiers
    (see the file grammar.peg)
    """

    RESERVED_KEYWORDS = [
        'contains',
        'icontains',
        'exact',
        'iexact',
    ]

    def __init__(self, default_field_lookups: list, min_search_characters: int):
        self.default_field_lookups = default_field_lookups
        self.min_search_characters = min_search_characters

    @staticmethod
    def process_value(field, value):
        # If all we're doing when we have a type mismatch with a field
        # is returning an empty set, then we don't need to do type validation.
        # Django compares between field values and string versions just fine.
        # But there's no magic string for null, so we're adding one.

        # TODO: Use Django or DRF machinery (or JSON parsing?) to handle types
        # that need special treatment, like dates

        # Handle None value
        if value == 'null':
            return None

        # Handle booleans - necessary when querying inside JSONBFields, and
        # also some other contexts: see `get_parsed_parameters()`
        try:
            lower_value = value.lower()
        except AttributeError:
            pass
        else:
            if lower_value in ['true', 'false']:
                return bool(util.strtobool(lower_value))

        return value

    @staticmethod
    def query(text, a, b, elements):
        exp = elements[1]
        if hasattr(exp, 'text') and exp.text == '':
            # Handle the empty query case with an empty Q object, returning all
            return Q()
        else:
            # fallthrough
            return exp

    @staticmethod
    def orexp(text, a, b, elements):
        # fallthrough if singular
        if elements[1].text == '':
            return elements[0]
        # else, combine full sequence of ORs into flattened expression
        else:
            # Start with the first Q object
            orgroup = elements[0]
            # Loop through the repeated clauses and OR the subexpressions.
            for clause in elements[1].elements:
                orgroup |= clause.expr
            return orgroup

    @staticmethod
    def andexp(text, a, b, elements):
        # fallthrough if singular
        if elements[1].text == '':
            return elements[0]
        # else, combine full sequence of ANDs into flattened expression
        else:
            # Start with the first Q object
            andgroup = elements[0]
            # Loop through the repeated clauses and AND the subexpressions.
            for clause in elements[1].elements:
                andgroup &= clause.expr
            return andgroup

    @staticmethod
    def parenexp(text, a, b, elements):
        # fallthrough to subexpression
        exp = elements[2]
        return exp

    @staticmethod
    def notexp(text, a, b, elements):
        # negate subexpression (Q object)
        exp = elements[2]
        return ~exp

    def term(self, text, a, b, elements):

        def _get_value(_field, _elements):
            # A search term by itself without a specified field
            _value = _elements[1]
            # the `field` value is not used in `process_value()`
            return self.process_value(_field, _value)

        if elements[0].text == '':
            value = _get_value('', elements)

            # As discussed here: https://github.com/kobotoolbox/kpi/pull/2830
            # there strain on the server for small search queries without a
            # specified field. The user will receive an empty list in response
            # until using `self.min_search_characters` or more characters
            if len(value) < self.min_search_characters:
                raise SearchQueryTooShortException()

            # A list of `Q` objects where every value is the same
            # searched value
            q_list = [
                Q(**{field: value}) for field in self.default_field_lookups
            ]
            # combining all the `Q` objects with an `or` operator and
            # returning the result
            return reduce(operator.or_, q_list)
        else:
            # A field+colon, and a value [[field,':'],value]
            field = elements[0].elements[0]

            # Bypass `status` field because it does not really exist.
            # It's only a property of Asset model.
            if field == 'status':
                return Q()

            # Trivial case, `field` is not a list inside a jsonb field.
            if not re.search(r'(\[\]$|\[\]__.+)', field):
                value = _get_value(field, elements)
                return Q(**{field: value})

            value = _get_value('', elements)
            field, parts = field.split('[]')

            if not value:  # search for an empty list
                return Q(**{f'{field}': []})

            if not parts:  # search within a list for an exact match
                return Q(**{f'{field}': [value]})

            dict_key, *field_lookup = parts.strip('__').split('__')

            if dict_key in self.RESERVED_KEYWORDS:
                if field_lookup:
                    raise QueryParserBadSyntax

                return self.get_q_for_list(
                    field, field_lookup=dict_key, value=value
                )

            return self.get_q_for_list(
                field, field_lookup=field_lookup, value=value, dict_key=dict_key
            )

    def get_q_for_list(self, field, field_lookup, value, dict_key=''):

        if dict_key:

            field_lookup = field_lookup[0] if field_lookup else None

            if field_lookup == 'exact':

                q_list = [
                    Q(**{f'{field}__contains': [{dict_key: value}]}),
                    Q(**{f'{field}__1__isnull': True}),
                ]
                return reduce(operator.and_, q_list)

            if not field_lookup or field_lookup == 'contains':

                if not value:  # search an empty value, same behaviour as an empty list
                    return Q(**{f'{field}': []})

                # contains: List should contain at least one dict which
                # equals `value`, e.g. '{"value": "CAN"}' should match one of
                # the items inside:
                # [
                #   {"value": "CAN", "label": "Canada"},
                #   {"value": "CAN", "label": "USA"}
                # ]
                return Q(**{f'{field}__contains': [{dict_key: value}]})
        else:

            if field_lookup == 'iexact':
                # List should contain only one item which equals `value`
                # but it is case-insensitive.
                q_list = [
                    Q(**{f'{field}__icontains': value}),
                    Q(**{f'{field}__1__isnull': True}),
                ]
                return reduce(operator.and_, q_list)

            if field_lookup in ['icontains', 'contains']:
                # icontains: List should contain at least one item which
                # matches part of `value`, e.g. "english" value will
                # work with "English (en)" in DB

                # contains: List should contain at least one item which
                # equals `value`, e.g. "English" value will not work
                # with "English (en)" in DB. `value` must be "English (en)"
                return Q(**{f'{field}__{field_lookup}': value})

        raise QueryParserNotSupportedFieldLookup

    @staticmethod
    def word(text, a, b, elements):
        return text[a:b]

    @staticmethod
    def string(text, a, b, elements):
        return text[a+1:b-1]

    @staticmethod
    def name(text, a, b, elements):
        return text[a:b]


def get_parsed_parameters(parsed_query: Q) -> dict:
    """
    NOTE: this is a hack that does not respect boolean logic.
    Returns a dictionary of all parameters detected in the query and their
    values. Values are always returned as list even if there is only one value
    found.
    For example:
    `q=parent_uid:foo AND asset_type:survey OR asset_type:block` returns
    {'parent_uid':['foo'], 'asset_type': ['survey', 'block']}

    """

    parameters = defaultdict(list)
    for child in parsed_query.__dict__['children']:
        if isinstance(child, Q):
            parameters.update(get_parsed_parameters(child))
            continue

        parameters[child[0]].append(child[1])

    # Cast to `dict` to be able to raise KeyError when accessing a non-existing
    # member of returned value
    return dict(parameters)


@cache_for_request
def parse(
    query: str,
    default_field_lookups: list,
    min_search_characters: int = None,
) -> Q:
    """
    Parse a Boolean query string into a Django Q object.
    If no field is specified in the query, `default_field_lookups` is assumed.
    For example, if `default_field_lookups` is a list containing
    `summary__icontains` and `name__icontains`, then the query `term` returns
    any object whose `summary` or `name` field contains `term` (case
    insensitive)
    """
    if not min_search_characters:
        min_search_characters = settings.MINIMUM_DEFAULT_SEARCH_CHARACTERS

    return grammar_parse(
        query,
        QueryParseActions(default_field_lookups, min_search_characters),
    )

1	# coding: utf-8
2	import operator	2✔
3	import re	2✔
4	from collections import defaultdict	2✔
5	from distutils import util	2✔
6	from functools import reduce	2✔
7
8	from django.conf import settings	2✔
9	from django.db.models import Q	2✔
10	from django_request_cache import cache_for_request	2✔
11
12	from kpi.exceptions import (	2✔
13	SearchQueryTooShortException,
14	QueryParserBadSyntax,
15	QueryParserNotSupportedFieldLookup,
16	)
17	from .canopy_autogenerated_parser import parse as grammar_parse	2✔
18
19
20	"""	1✔
21	This is a utility for parsing a Boolean, Whoosh-like query string and
22	translating it into a Django Q object, which can then be used to filter a
23	queryset in the ORM.
24
25	Syntax examples:
26	* `name:term` returns any object whose `name` field exactly matches
27	`term` (case sensitive)
28	* `owner__username=meg` traverses the `owner` relationship, returning
29	any object whose `owner` field matches an object whose `username` field
30	exactly matches `meg` (case sensitive)
31	* `color:orange NOT (type:fruit OR type:chair)` returns anything
32	whose color is orange so long as it is not a fruit or a chair. AND, OR,
33	and NOT operators are supported. They must be in ALL CAPS.
34
35	Special notes:
36	* The value `null` in a query is translated to `None`, e.g. `parent:null`
37	effectively becomes the ORM `filter(parent=None)`
38	"""
39
40
41	class QueryParseActions:	2✔
42	"""
43	Actions for the parser to take when it encounters certain identifiers
44	(see the file grammar.peg)
45	"""
46
47	RESERVED_KEYWORDS = [	2✔
48	'contains',
49	'icontains',
50	'exact',
51	'iexact',
52	]
53
54	def __init__(self, default_field_lookups: list, min_search_characters: int):	2✔
55	self.default_field_lookups = default_field_lookups	2✔
56	self.min_search_characters = min_search_characters	2✔
57
58	@staticmethod	2✔
59	def process_value(field, value):	2✔
60	# If all we're doing when we have a type mismatch with a field
61	# is returning an empty set, then we don't need to do type validation.
62	# Django compares between field values and string versions just fine.
63	# But there's no magic string for null, so we're adding one.
64
65	# TODO: Use Django or DRF machinery (or JSON parsing?) to handle types
66	# that need special treatment, like dates
67
68	# Handle None value
69	if value == 'null':	2✔
70	return None	2✔
71
72	# Handle booleans - necessary when querying inside JSONBFields, and
73	# also some other contexts: see `get_parsed_parameters()`
74	try:	2✔
75	lower_value = value.lower()	2✔
76	except AttributeError:	×
77	pass	×
78	else:
79	if lower_value in ['true', 'false']:	2!
80	return bool(util.strtobool(lower_value))	×
81
82	return value	2✔
83
84	@staticmethod	2✔
85	def query(text, a, b, elements):	2✔
86	exp = elements[1]	2✔
87	if hasattr(exp, 'text') and exp.text == '':	2!
88	# Handle the empty query case with an empty Q object, returning all
89	return Q()	×
90	else:
91	# fallthrough
92	return exp	2✔
93
94	@staticmethod	2✔
95	def orexp(text, a, b, elements):	2✔
96	# fallthrough if singular
97	if elements[1].text == '':	2✔
98	return elements[0]	2✔
99	# else, combine full sequence of ORs into flattened expression
100	else:
101	# Start with the first Q object
102	orgroup = elements[0]	2✔
103	# Loop through the repeated clauses and OR the subexpressions.
104	for clause in elements[1].elements:	2✔
105	orgroup \|= clause.expr	2✔
106	return orgroup	2✔
107
108	@staticmethod	2✔
109	def andexp(text, a, b, elements):	2✔
110	# fallthrough if singular
111	if elements[1].text == '':	2✔
112	return elements[0]	2✔
113	# else, combine full sequence of ANDs into flattened expression
114	else:
115	# Start with the first Q object
116	andgroup = elements[0]	2✔
117	# Loop through the repeated clauses and AND the subexpressions.
118	for clause in elements[1].elements:	2✔
119	andgroup &= clause.expr	2✔
120	return andgroup	2✔
121
122	@staticmethod	2✔
123	def parenexp(text, a, b, elements):	2✔
124	# fallthrough to subexpression
125	exp = elements[2]	2✔
126	return exp	2✔
127
128	@staticmethod	2✔
129	def notexp(text, a, b, elements):	2✔
130	# negate subexpression (Q object)
131	exp = elements[2]	2✔
132	return ~exp	2✔
133
134	def term(self, text, a, b, elements):	2✔
135
136	def _get_value(_field, _elements):	2✔
137	# A search term by itself without a specified field
138	_value = _elements[1]	2✔
139	# the `field` value is not used in `process_value()`
140	return self.process_value(_field, _value)	2✔
141
142	if elements[0].text == '':	2✔
143	value = _get_value('', elements)	2✔
144
145	# As discussed here: https://github.com/kobotoolbox/kpi/pull/2830
146	# there strain on the server for small search queries without a
147	# specified field. The user will receive an empty list in response
148	# until using `self.min_search_characters` or more characters
149	if len(value) < self.min_search_characters:	2✔
150	raise SearchQueryTooShortException()	2✔
151
152	# A list of `Q` objects where every value is the same
153	# searched value
154	q_list = [	2✔
155	Q(**{field: value}) for field in self.default_field_lookups
156	]
157	# combining all the `Q` objects with an `or` operator and
158	# returning the result
159	return reduce(operator.or_, q_list)	2✔
160	else:
161	# A field+colon, and a value [[field,':'],value]
162	field = elements[0].elements[0]	2✔
163
164	# Bypass `status` field because it does not really exist.
165	# It's only a property of Asset model.
166	if field == 'status':	2!
167	return Q()	×
168
169	# Trivial case, `field` is not a list inside a jsonb field.
170	if not re.search(r'(\[\]$\|\[\]__.+)', field):	2✔
171	value = _get_value(field, elements)	2✔
172	return Q(**{field: value})	2✔
173
174	value = _get_value('', elements)	2✔
175	field, parts = field.split('[]')	2✔
176
177	if not value: # search for an empty list	2✔
178	return Q(**{f'{field}': []})	2✔
179
180	if not parts: # search within a list for an exact match	2✔
181	return Q(**{f'{field}': [value]})	2✔
182
183	dict_key, *field_lookup = parts.strip('__').split('__')	2✔
184
185	if dict_key in self.RESERVED_KEYWORDS:	2!
186	if field_lookup:	×
187	raise QueryParserBadSyntax	×
188
189	return self.get_q_for_list(	×
190	field, field_lookup=dict_key, value=value
191	)
192
193	return self.get_q_for_list(	2✔
194	field, field_lookup=field_lookup, value=value, dict_key=dict_key
195	)
196
197	def get_q_for_list(self, field, field_lookup, value, dict_key=''):	2✔
198
199	if dict_key:	2!
200
201	field_lookup = field_lookup[0] if field_lookup else None	2✔
202
203	if field_lookup == 'exact':	2!
204
205	q_list = [	×
206	Q(**{f'{field}__contains': [{dict_key: value}]}),
207	Q(**{f'{field}__1__isnull': True}),
208	]
209	return reduce(operator.and_, q_list)	×
210
211	if not field_lookup or field_lookup == 'contains':	2✔
212
213	if not value: # search an empty value, same behaviour as an empty list	2!
214	return Q(**{f'{field}': []})	×
215
216	# contains: List should contain at least one dict which
217	# equals `value`, e.g. '{"value": "CAN"}' should match one of
218	# the items inside:
219	# [
220	# {"value": "CAN", "label": "Canada"},
221	# {"value": "CAN", "label": "USA"}
222	# ]
223	return Q(**{f'{field}__contains': [{dict_key: value}]})	2✔
224	else:
225
226	if field_lookup == 'iexact':	×
227	# List should contain only one item which equals `value`
228	# but it is case-insensitive.
229	q_list = [	×
230	Q(**{f'{field}__icontains': value}),
231	Q(**{f'{field}__1__isnull': True}),
232	]
233	return reduce(operator.and_, q_list)	×
234
235	if field_lookup in ['icontains', 'contains']:	×
236	# icontains: List should contain at least one item which
237	# matches part of `value`, e.g. "english" value will
238	# work with "English (en)" in DB
239
240	# contains: List should contain at least one item which
241	# equals `value`, e.g. "English" value will not work
242	# with "English (en)" in DB. `value` must be "English (en)"
243	return Q(**{f'{field}__{field_lookup}': value})	×
244
245	raise QueryParserNotSupportedFieldLookup	2✔
246
247	@staticmethod	2✔
248	def word(text, a, b, elements):	2✔
249	return text[a:b]	2✔
250
251	@staticmethod	2✔
252	def string(text, a, b, elements):	2✔
253	return text[a+1:b-1]	2✔
254
255	@staticmethod	2✔
256	def name(text, a, b, elements):	2✔
257	return text[a:b]	2✔
258
259
260	def get_parsed_parameters(parsed_query: Q) -> dict:	2✔
261	"""
262	NOTE: this is a hack that does not respect boolean logic.
263	Returns a dictionary of all parameters detected in the query and their
264	values. Values are always returned as list even if there is only one value
265	found.
266	For example:
267	`q=parent_uid:foo AND asset_type:survey OR asset_type:block` returns
268	{'parent_uid':['foo'], 'asset_type': ['survey', 'block']}
269
270	"""
271
272	parameters = defaultdict(list)	2✔
273	for child in parsed_query.__dict__['children']:	2✔
274	if isinstance(child, Q):	2!
275	parameters.update(get_parsed_parameters(child))	×
276	continue	×
277
278	parameters[child[0]].append(child[1])	2✔
279
280	# Cast to `dict` to be able to raise KeyError when accessing a non-existing
281	# member of returned value
282	return dict(parameters)	2✔
283
284
285	@cache_for_request	2✔
286	def parse(	2✔
287	query: str,
288	default_field_lookups: list,
289	min_search_characters: int = None,
290	) -> Q:
291	"""
292	Parse a Boolean query string into a Django Q object.
293	If no field is specified in the query, `default_field_lookups` is assumed.
294	For example, if `default_field_lookups` is a list containing
295	`summary__icontains` and `name__icontains`, then the query `term` returns
296	any object whose `summary` or `name` field contains `term` (case
297	insensitive)
298	"""
299	if not min_search_characters:	2✔
300	min_search_characters = settings.MINIMUM_DEFAULT_SEARCH_CHARACTERS	2✔
301
302	return grammar_parse(	2✔
303	query,
304	QueryParseActions(default_field_lookups, min_search_characters),
305	)

kobotoolbox / kpi / 3683952713

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous