Coverage for app/venv/lib/python3.14/site-packages/weblate/utils/search.py: 24%
424 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 07:15 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 07:15 +0000
1# Copyright © Michal Čihař <michal@weblate.org>
2#
3# SPDX-License-Identifier: GPL-3.0-or-later
5from __future__ import annotations
7import threading
8import warnings
9from datetime import datetime
10from functools import lru_cache, reduce
11from itertools import chain
12from operator import and_, or_
13from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast, overload
15from dateutil.parser import ParserError
16from dateutil.parser import parse as dateutil_parse
17from django.db import transaction
18from django.db.models import Count, F, Q, Value
19from django.db.utils import DataError, OperationalError
20from django.http import Http404
21from django.utils import timezone
22from django.utils.translation import gettext
23from pyparsing import (
24 CaselessKeyword,
25 OpAssoc,
26 Optional,
27 ParseResults,
28 Regex,
29 Word,
30 alphanums,
31 infix_notation,
32 one_of,
33)
35from weblate.checks.parser import RawQuotedString
36from weblate.lang.models import Language
37from weblate.trans.models import Category, Component, Project, Translation
38from weblate.trans.util import PLURAL_SEPARATOR
39from weblate.utils.db import re_escape, using_postgresql
40from weblate.utils.state import (
41 STATE_APPROVED,
42 STATE_FUZZY,
43 STATE_NAMES,
44 STATE_READONLY,
45 STATE_TRANSLATED,
46)
47from weblate.utils.stats import CategoryLanguage, ProjectLanguage
48from weblate.utils.views import parse_path
50if TYPE_CHECKING: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true
51 from collections.abc import Callable
53 from django.db.models import Expression
54 from pyparsing import (
55 ParserElement,
56 )
59# Helper parsing objects
60class RegexExpr:
61 def __init__(self, tokens) -> None:
62 self.expr = tokens[1]
65class RangeExpr:
66 def __init__(self, tokens) -> None:
67 self.start = tokens[1]
68 self.end = tokens[3]
71OPERATOR_MAP = {
72 ":": "substring",
73 ":=": "exact",
74 ":<": "lt",
75 ":<=": "lte",
76 ":>": "gt",
77 ":>=": "gte",
78}
81def build_parser(term_expression: type[BaseTermExpr]) -> ParserElement:
82 """Build parsing grammar."""
83 # Booleans
84 op_and = CaselessKeyword("AND")
85 op_or = Optional(CaselessKeyword("OR"))
86 op_not = CaselessKeyword("NOT")
88 # Search operator
89 operator = one_of(OPERATOR_MAP.keys())
91 # Field name, explicitly exclude URL like patterns
92 field = Regex(r"""(?!http|ftp|https|mailto)[a-zA-Z_]+""")
94 # Match token
95 word = Regex(r"""[^ \r\n\(\)]([^ \r\n'"]*[^ \r\n'"\)])?""")
96 date = Word(alphanums + ":._-")
98 # Date range
99 date_range = "[" + date + "to" + date + "]"
100 date_range.add_parse_action(RangeExpr)
102 # Match value
103 regex_string = "r" + RawQuotedString('"')
104 regex_string.add_parse_action(RegexExpr)
105 string = regex_string | RawQuotedString("'") | RawQuotedString('"') | word
107 # Single term, either field specific or not
108 term = (field + operator + (date_range | string)) | string
109 term.add_parse_action(term_expression)
111 # Multi term with or without operator
112 return Optional(
113 infix_notation(
114 term,
115 [
116 (
117 op_not,
118 1,
119 OpAssoc.RIGHT,
120 ),
121 (
122 op_and,
123 2,
124 OpAssoc.LEFT,
125 ),
126 (
127 op_or,
128 2,
129 OpAssoc.LEFT,
130 ),
131 ],
132 )
133 )
136class BaseTermExpr:
137 PLAIN_FIELDS: ClassVar[set[str]] = set()
138 NONTEXT_FIELDS: ClassVar[dict[str, str]] = {}
139 STRING_FIELD_MAP: ClassVar[dict[str, str]] = {}
140 EXACT_FIELD_MAP: ClassVar[dict[str, str]] = {}
141 enable_fulltext: ClassVar[bool] = True
143 def __init__(self, tokens) -> None:
144 if len(tokens) == 1:
145 self.field = None
146 self.operator = ":"
147 self.match = tokens[0]
148 else:
149 self.field, self.operator, self.match = tokens
150 self.fixup()
152 def __repr__(self) -> str:
153 return f"<{self.__class__.__name__}: {self.field!r}, {self.operator!r}, {self.match!r}>"
155 def fixup(self) -> None:
156 # Avoid unwanted lt/gt searches on plain text fields
157 if self.field in self.PLAIN_FIELDS and self.operator not in {":", ":="}:
158 self.match = f"{self.operator[1:]}{self.match}"
159 self.operator = ":"
161 def convert_state(self, text: str) -> int | None:
162 if text is None:
163 return None
164 if text.isdigit():
165 return int(text)
166 try:
167 return STATE_NAMES[text]
168 except KeyError as exc:
169 raise ValueError(gettext("Unsupported state: {}").format(text)) from exc
171 def convert_bool(self, text: str) -> bool:
172 ltext = text.lower()
173 if ltext in {"yes", "true", "on", "1"}:
174 return True
175 if ltext in {"no", "false", "off", "0"}:
176 return False
177 msg = f"Invalid boolean value: {text}"
178 raise ValueError(msg)
180 @overload
181 def convert_int(self, text: RangeExpr) -> tuple[int, int]: ... 181 ↛ anywhereline 181 didn't jump anywhere: it always raised an exception.
182 @overload
183 def convert_int(self, text: str) -> int: ... 183 ↛ anywhereline 183 didn't jump anywhere: it always raised an exception.
184 def convert_int(self, text):
185 if isinstance(text, RangeExpr):
186 return (
187 self.convert_int(text.start),
188 self.convert_int(text.end),
189 )
190 return int(text)
192 def convert_id(self, text: str) -> int | set[int]:
193 if "," in text:
194 return {self.convert_int(part) for part in text.split(",")}
195 return self.convert_int(text)
197 @overload
198 def convert_datetime( 198 ↛ anywhereline 198 didn't jump anywhere: it always raised an exception.
199 self,
200 text: RangeExpr,
201 ) -> tuple[datetime, datetime]: ...
202 @overload
203 def convert_datetime( 203 ↛ anywhereline 203 didn't jump anywhere: it always raised an exception.
204 self,
205 text: str,
206 ) -> datetime: ...
207 def convert_datetime(self, text):
208 if isinstance(text, RangeExpr):
209 return (
210 self.date_parse(text.start, hour=0, minute=0, second=0, microsecond=0),
211 self.date_parse(
212 text.end, hour=23, minute=59, second=59, microsecond=999999
213 ),
214 )
215 if text.isdigit() and len(text) == 4:
216 tzinfo = timezone.get_current_timezone()
217 year = int(text)
218 return (
219 datetime(
220 year=year,
221 month=1,
222 day=1,
223 hour=0,
224 minute=0,
225 second=0,
226 microsecond=0,
227 tzinfo=tzinfo,
228 ),
229 datetime(
230 year=year,
231 month=12,
232 day=31,
233 hour=23,
234 minute=59,
235 second=59,
236 microsecond=999999,
237 tzinfo=tzinfo,
238 ),
239 )
241 return self.date_parse(text)
243 def get_day_range(self, timestamp: datetime) -> tuple[datetime, datetime]:
244 return (
245 timestamp.replace(hour=0, minute=0, second=0, microsecond=0),
246 timestamp.replace(hour=23, minute=59, second=59, microsecond=999999),
247 )
249 @overload
250 def date_parse_human( 250 ↛ anywhereline 250 didn't jump anywhere: it always raised an exception.
251 self,
252 text: str,
253 hour: None = None,
254 minute: None = None,
255 second: None = None,
256 microsecond: None = None,
257 ) -> datetime | tuple[datetime, datetime]: ...
258 @overload
259 def date_parse_human( 259 ↛ anywhereline 259 didn't jump anywhere: it always raised an exception.
260 self,
261 text: str,
262 hour: int,
263 minute: int,
264 second: int,
265 microsecond: int,
266 ) -> datetime: ...
267 def date_parse_human(
268 self,
269 text,
270 hour=None,
271 minute=None,
272 second=None,
273 microsecond=None,
274 ):
275 # Lazily import as this can be expensive
276 from dateparser.date import DateDataParser
278 # Custom RELATIVE_BASE allows to base "1 day ago" from the midnight instead
279 # of the current time
280 parser = DateDataParser(
281 locales=["en"],
282 settings={
283 "RELATIVE_BASE": timezone.now().replace(
284 hour=0, minute=0, second=0, microsecond=0
285 )
286 },
287 )
289 # Attempts to parse the text using dateparser
290 # If the text is unparsable it will return None
291 data = parser.get_date_data(text)
293 date_obj: datetime | None = data.date_obj
295 if date_obj is None:
296 msg = "Could not parse timestamp"
297 raise ValueError(msg)
299 # Always include timezone
300 if date_obj.tzinfo is None:
301 date_obj = date_obj.replace(tzinfo=timezone.get_current_timezone())
303 if data.period == "day":
304 if (
305 hour is not None
306 and minute is not None
307 and second is not None
308 and microsecond is not None
309 ):
310 # Replace timestamp when parsing range
311 return date_obj.replace(
312 hour=hour,
313 minute=minute,
314 second=second,
315 microsecond=microsecond,
316 tzinfo=timezone.get_current_timezone(),
317 )
318 # Create one day range from date only
319 if (
320 date_obj.hour == 0
321 and date_obj.minute == 0
322 and date_obj.second == 0
323 and date_obj.microsecond == 0
324 ):
325 return self.get_day_range(date_obj)
327 return date_obj
329 @overload
330 def date_parse( 330 ↛ anywhereline 330 didn't jump anywhere: it always raised an exception.
331 self,
332 text: str,
333 hour: None = None,
334 minute: None = None,
335 second: None = None,
336 microsecond: None = None,
337 ) -> datetime | tuple[datetime, datetime]: ...
338 @overload
339 def date_parse( 339 ↛ anywhereline 339 didn't jump anywhere: it always raised an exception.
340 self,
341 text: str,
342 hour: int,
343 minute: int,
344 second: int,
345 microsecond: int,
346 ) -> datetime: ...
347 def date_parse(
348 self,
349 text,
350 hour=None,
351 minute=None,
352 second=None,
353 microsecond=None,
354 ):
355 result: datetime | None
357 default = timezone.now()
358 if hour is None or minute is None or second is None or microsecond is None:
359 # Here we inject 5:55:55 time and if that was not changed
360 # during parsing, we assume it was not specified while
361 # generating the query
362 default = default.replace(hour=5, minute=5, second=5, microsecond=5)
363 else:
364 # Apply real defaults
365 default = default.replace(
366 hour=hour, minute=minute, second=second, microsecond=microsecond
367 )
369 try:
370 with warnings.catch_warnings():
371 # Ignore ambiguous date warning, it is gracefully handled by datetutil
372 # or raises exception.
373 warnings.filterwarnings(
374 "ignore",
375 "Parsing dates involving a day of month without a year specified",
376 DeprecationWarning,
377 )
378 result = dateutil_parse(text, default=default)
379 except ParserError:
380 result = None
382 if not result:
383 return self.date_parse_human(
384 text, hour=hour, minute=minute, second=second, microsecond=microsecond
385 )
386 msg = "Could not parse timestamp"
387 raise ValueError(msg)
389 if (
390 hour is None
391 and result.hour == 5
392 and result.minute == 5
393 and result.second == 5
394 and result.microsecond == 5
395 ):
396 return (
397 result.replace(hour=0, minute=0, second=0, microsecond=0),
398 result.replace(hour=23, minute=59, second=59, microsecond=999999),
399 )
401 return result
403 def convert_change_action(self, text: str) -> int:
404 from weblate.trans.models import Change
406 try:
407 return Change.ACTION_NAMES[text]
408 except KeyError:
409 return Change.ACTION_STRINGS[text]
411 def convert_change_time(self, text: str) -> datetime | tuple[datetime, datetime]:
412 return self.convert_datetime(text)
414 def field_name(self, field: str, suffix: str | None = None) -> str:
415 if suffix is None:
416 suffix = OPERATOR_MAP[self.operator]
418 if field in self.EXACT_FIELD_MAP:
419 # Change contains to exact, do not change other (for example regex)
420 if suffix == "substring":
421 suffix = "iexact"
422 return f"{self.EXACT_FIELD_MAP[field]}__{suffix}"
424 if not self.enable_fulltext and suffix == "substring":
425 suffix = "icontains"
427 if field in self.PLAIN_FIELDS:
428 return f"{field}__{suffix}"
429 if field in self.STRING_FIELD_MAP:
430 return f"{self.STRING_FIELD_MAP[field]}__{suffix}"
431 if field in self.NONTEXT_FIELDS:
432 if suffix not in {"substring", "iexact"}:
433 return f"{self.NONTEXT_FIELDS[field]}__{suffix}"
434 return self.NONTEXT_FIELDS[field]
435 msg = f"Unsupported field: {field}"
436 raise ValueError(msg)
438 def convert_non_field(self) -> Q:
439 raise NotImplementedError
441 def as_query(self, context: dict) -> Q:
442 field = self.field
443 match = self.match
444 # Simple term based search
445 if not field:
446 return self.convert_non_field()
448 # Field specific code
449 field_method: Callable[[str, dict], Q] | None = cast(
450 "Callable[[str, dict], Q] | None",
451 getattr(self, f"{field}_field", None),
452 )
453 if field_method is not None:
454 return field_method(match, context)
456 # Field conversion
457 convert_method = getattr(self, f"convert_{field}", None)
458 if convert_method is not None:
459 match = convert_method(match)
461 if isinstance(match, RegexExpr):
462 # Regular expression
463 from weblate.trans.models import Unit
465 with transaction.atomic():
466 try:
467 Unit.objects.annotate(test=Value("")).filter(
468 test__trgm_regex=match.expr
469 ).exists()
470 except (DataError, OperationalError) as error:
471 # PostgreSQL raises DataError, MySQL OperationalError
472 raise ValueError(
473 gettext("Invalid regular expression: {}").format(error)
474 ) from error
475 return Q(**{self.field_name(field, "trgm_regex"): match.expr})
477 if isinstance(match, tuple):
478 start, end = match
479 # Ranges
480 if self.operator in {":", ":="}:
481 query = Q(**{self.field_name(field, "range"): (start, end)})
482 elif self.operator in {":>", ":>="}:
483 query = Q(**{self.field_name(field, "gte"): start})
484 else:
485 query = Q(**{self.field_name(field, "lte"): end})
487 elif isinstance(match, set):
488 query = Q(**{self.field_name(field, "in"): match})
489 else:
490 # Generic query
491 query = Q(**{self.field_name(field): match})
493 return self.field_extra(field, query, match)
495 def get_annotations(self, context: dict) -> dict[str, Expression]:
496 return {}
498 def field_extra(self, field: str, query: Q, match: Any) -> Q: # noqa: ANN401
499 return query
501 def is_field(self, text: str, context: dict) -> Q:
502 msg = f"Unsupported is lookup: {text}"
503 raise ValueError(msg)
505 def has_field(self, text: str, context: dict) -> Q:
506 msg = f"Unsupported has lookup: {text}"
507 raise ValueError(msg)
510class UnitTermExpr(BaseTermExpr):
511 PLAIN_FIELDS: ClassVar[set[str]] = {
512 "source",
513 "target",
514 "context",
515 "note",
516 "location",
517 }
518 NONTEXT_FIELDS: ClassVar[dict[str, str]] = {
519 "priority": "priority",
520 "id": "id",
521 "state": "state",
522 "source_state": "source_unit__state",
523 "position": "position",
524 "pending": "pending_changes__isnull",
525 "changed": "change__timestamp",
526 "source_changed": "source_unit__last_updated",
527 "last_changed": "last_updated",
528 "change_time": "change__timestamp",
529 "added": "timestamp",
530 "change_action": "change__action",
531 "labels_count": "labels_count",
532 }
533 STRING_FIELD_MAP: ClassVar[dict[str, str]] = {
534 "suggestion": "suggestion__target",
535 "comment": "comment__comment",
536 "resolved_comment": "comment__comment",
537 "key": "context",
538 "explanation": "source_unit__explanation",
539 }
540 EXACT_FIELD_MAP: ClassVar[dict[str, str]] = {
541 "check": "check__name",
542 "dismissed_check": "check__name",
543 "language": "translation__language__code",
544 "project": "translation__component__project__slug",
545 "changed_by": "change__author__username",
546 "suggestion_author": "suggestion__user__username",
547 "comment_author": "comment__user__username",
548 "label": "source_unit__labels__name",
549 "screenshot": "source_unit__screenshots__name",
550 }
552 def is_field(self, text: str, context: dict) -> Q:
553 if text in {"read-only", "readonly"}:
554 return Q(state=STATE_READONLY)
555 if text == "approved":
556 return Q(state=STATE_APPROVED)
557 if text in {"fuzzy", "needs-editing"}:
558 return Q(state=STATE_FUZZY)
559 if text == "translated":
560 return Q(state__gte=STATE_TRANSLATED)
561 if text == "untranslated":
562 return Q(state__lt=STATE_TRANSLATED)
563 if text == "pending":
564 return Q(pending_changes__isnull=False)
566 return super().is_field(text, context)
568 def has_field(self, text: str, context: dict) -> Q: # noqa: C901
569 if text == "plural":
570 return Q(source__search=PLURAL_SEPARATOR)
571 if text == "suggestion":
572 return Q(suggestion__isnull=False)
573 if text == "explanation":
574 return ~Q(source_unit__explanation="")
575 if text == "location":
576 return ~Q(location="")
577 if text == "note":
578 return ~Q(note="")
579 if text == "comment":
580 return Q(comment__resolved=False)
581 if text in {"resolved-comment", "resolved_comment"}:
582 return Q(comment__resolved=True)
583 if text in {"check", "failing-check", "failing_check"}:
584 return Q(check__dismissed=False)
585 if text in {
586 "dismissed-check",
587 "dismissed_check",
588 "ignored-check",
589 "ignored_check",
590 }:
591 return Q(check__dismissed=True)
592 if text == "translation":
593 return Q(state__gte=STATE_TRANSLATED)
594 if text in {"variant", "shaping"}:
595 return Q(defined_variants__isnull=False) | (
596 ~Q(variant__variant_regex="")
597 & Q(context__regex=F("variant__variant_regex"))
598 )
599 if text == "label":
600 return Q(source_unit__labels__isnull=False) | Q(labels__isnull=False)
601 if text == "context":
602 return ~Q(context="")
603 if text == "screenshot":
604 return Q(screenshots__isnull=False) | Q(
605 source_unit__screenshots__isnull=False
606 )
607 if text == "flags":
608 return ~Q(source_unit__extra_flags="")
609 if text == "glossary":
610 project = context.get("project")
611 if not project:
612 return Q(source__isnull=True)
613 terms = set(
614 chain.from_iterable(
615 glossary.glossary_sources for glossary in project.glossaries
616 )
617 )
618 if not terms:
619 return Q(source__isnull=True)
620 if using_postgresql():
621 template = r"[[:<:]]({})[[:>:]]"
622 else:
623 template = r"(^|[ \t\n\r\f\v])({})($|[ \t\n\r\f\v])"
624 return Q(
625 source__iregex=template.format(
626 "|".join(re_escape(term) for term in terms)
627 )
628 )
630 return super().has_field(text, context)
632 def convert_source_state(self, text: str) -> int | None:
633 return self.convert_state(text)
635 def component_field(self, text: str, context: dict) -> Q:
636 if self.operator == ":=":
637 return Q(translation__component__slug__iexact=text) | Q(
638 translation__component__name__iexact=text
639 )
640 return Q(translation__component__slug__icontains=text) | Q(
641 translation__component__name__icontains=text
642 )
644 def path_field(self, text: str, context: dict) -> Q:
645 try:
646 obj = parse_path(
647 None,
648 text.split("/"),
649 (
650 Translation,
651 Component,
652 Project,
653 ProjectLanguage,
654 Category,
655 CategoryLanguage,
656 Language,
657 ),
658 )
659 except Http404:
660 return Q(translation=None)
662 if isinstance(obj, Translation):
663 return Q(translation=obj)
664 if isinstance(obj, Component):
665 return Q(translation__component=obj)
666 if isinstance(obj, Project):
667 return Q(translation__component__project=obj)
668 if isinstance(obj, ProjectLanguage):
669 return Q(translation__component__project=obj.project) & Q(
670 translation__language=obj.language
671 )
672 if isinstance(obj, Category):
673 return Q(translation__component_id__in=obj.all_component_ids)
674 if isinstance(obj, CategoryLanguage):
675 return Q(translation__component_id__in=obj.category.all_component_ids) & Q(
676 translation__language=obj.language
677 )
678 if isinstance(obj, Language):
679 return Q(translation__language=obj)
680 msg = f"Unsupported path lookup: {obj}"
681 raise TypeError(msg)
683 def pending_field(self, text: str, context: dict) -> Q:
684 boolean_value = self.convert_bool(text)
685 return Q(pending_changes__isnull=not boolean_value)
687 def convert_changed(self, text: str) -> datetime | tuple[datetime, datetime]:
688 return self.convert_datetime(text)
690 def convert_source_changed(self, text: str) -> datetime | tuple[datetime, datetime]:
691 return self.convert_datetime(text)
693 def convert_last_changed(self, text: str) -> datetime | tuple[datetime, datetime]:
694 return self.convert_datetime(text)
696 def convert_added(self, text: str) -> datetime | tuple[datetime, datetime]:
697 return self.convert_datetime(text)
699 def convert_position(self, text: str) -> int:
700 return self.convert_int(text)
702 def convert_priority(self, text: str) -> int:
703 return self.convert_int(text)
705 def convert_labels_count(self, text: str) -> int:
706 return self.convert_int(text)
708 def field_extra(self, field: str, query: Q, match: Any) -> Q: # noqa: ANN401
709 from weblate.trans.models import Change
711 if field in {"changed", "changed_by"}:
712 return query & Q(change__action__in=Change.ACTIONS_CONTENT)
713 if field == "check":
714 return query & Q(check__dismissed=False)
715 if field == "dismissed_check":
716 return query & Q(check__dismissed=True)
717 if field == "label":
718 return query | Q(labels__name__iexact=match)
719 if field == "screenshot":
720 return query | Q(screenshots__name__iexact=match)
721 if field == "comment":
722 return query & Q(comment__resolved=False)
723 if field == "resolved_comment":
724 return query & Q(comment__resolved=True)
726 return super().field_extra(field, query, match)
728 def convert_non_field(self) -> Q:
729 return (
730 Q(source__substring=self.match)
731 | Q(target__substring=self.match)
732 | Q(context__substring=self.match)
733 )
735 def get_annotations(self, context: dict) -> dict[str, Expression]:
736 if self.field == "labels_count":
737 return {"labels_count": Count("source_unit__labels") + Count("labels")}
738 return super().get_annotations(context)
741class UserTermExpr(BaseTermExpr):
742 PLAIN_FIELDS: ClassVar[set[str]] = {"username", "full_name"}
743 NONTEXT_FIELDS: ClassVar[dict[str, str]] = {
744 "joined": "date_joined",
745 "change_time": "change__timestamp",
746 "change_action": "change__action",
747 }
748 EXACT_FIELD_MAP: ClassVar[dict[str, str]] = {
749 "language": "profile__languages__code",
750 "translates": "change__language__code",
751 }
752 enable_fulltext: ClassVar[bool] = False
754 def convert_joined(self, text: str) -> datetime | tuple[datetime, datetime]:
755 return self.convert_datetime(text)
757 def convert_non_field(self) -> Q:
758 return Q(username__icontains=self.match) | Q(full_name__icontains=self.match)
760 def contributes_field(self, text: str, context: dict) -> Q:
761 from weblate.trans.models import Component
763 if "/" not in text:
764 return Q(change__project__slug__iexact=text)
765 return Q(
766 change__component_id__in=list(
767 Component.objects.filter_by_path(text).values_list("id", flat=True)
768 )
769 )
772class SuperuserUserTermExpr(UserTermExpr):
773 STRING_FIELD_MAP: ClassVar[dict[str, str]] = {
774 "email": "social_auth__verifiedemail__email",
775 }
777 def convert_non_field(self) -> Q:
778 return (
779 Q(username__icontains=self.match)
780 | Q(full_name__icontains=self.match)
781 | Q(social_auth__verifiedemail__email__iexact=self.match)
782 )
784 def is_field(self, text: str, context: dict) -> Q:
785 if text == "active":
786 return Q(is_active=True)
787 if text == "bot":
788 return Q(is_bot=True)
789 if text == "superuser":
790 return Q(is_superuser=True)
792 return super().is_field(text, context)
795PARSERS: dict[Literal["unit", "user", "superuser"], ParserElement] = {
796 "unit": build_parser(UnitTermExpr),
797 "user": build_parser(UserTermExpr),
798 "superuser": build_parser(SuperuserUserTermExpr),
799}
800PARSER_LOCK = threading.Lock()
803def parser_to_query(obj: ParseResults | BaseTermExpr, context: dict) -> Q:
804 # Simple lookups
805 if isinstance(obj, BaseTermExpr): 805 ↛ 806line 805 didn't jump to line 806 because the condition on line 805 was never true
806 return obj.as_query(context)
808 # Operators
809 operator = ""
810 expressions: list[Q] = []
811 was_operator = False
812 for item in obj: 812 ↛ 813line 812 didn't jump to line 813 because the loop on line 812 never started
813 if isinstance(item, str) and (current := item.upper()) in {"OR", "AND", "NOT"}:
814 if operator and current != operator:
815 msg = "Mixed operators!"
816 raise ValueError(msg)
817 operator = current
818 was_operator = True
819 continue
820 if not was_operator and expressions:
821 # Implicit AND
822 expressions[-1] &= parser_to_query(item, context)
823 else:
824 expressions.append(parser_to_query(item, context))
825 was_operator = False
827 if not expressions: 827 ↛ 830line 827 didn't jump to line 830 because the condition on line 827 was always true
828 return Q()
830 if operator == "NOT":
831 return ~expressions[0]
832 if operator == "AND":
833 return reduce(and_, expressions)
834 return reduce(or_, expressions)
837def parser_annotations(
838 obj: ParseResults | BaseTermExpr, context: dict
839) -> dict[str, Expression]:
840 result: dict[str, Expression] = {}
841 if isinstance(obj, BaseTermExpr): 841 ↛ 842line 841 didn't jump to line 842 because the condition on line 841 was never true
842 result.update(obj.get_annotations(context))
843 else:
844 for item in obj: 844 ↛ 845line 844 didn't jump to line 845 because the loop on line 844 never started
845 if isinstance(item, (BaseTermExpr, ParseResults)):
846 result.update(parser_annotations(item, context))
847 return result
850@lru_cache(maxsize=32)
851def parse_string(
852 text: str, parser: Literal["unit", "user", "superuser"]
853) -> ParseResults:
854 if "\x00" in text: 854 ↛ 855line 854 didn't jump to line 855 because the condition on line 854 was never true
855 msg = "Invalid query string."
856 raise ValueError(msg)
857 with PARSER_LOCK:
858 return PARSERS[parser].parse_string(text, parse_all=True)
861def parse_query(
862 text: str, parser: Literal["unit", "user", "superuser"] = "unit", **context
863) -> tuple[Q, dict[str, Expression]]:
864 parsed = parse_string(text, parser)
865 return parser_to_query(parsed, context), parser_annotations(parsed, context)