1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
from typing import Optional
from moto.utilities.tokenizer import GenericTokenizer
class ParsedQuery:
def __init__(self) -> None:
self.limit: Optional[int] = None
self.fields: list[str] = []
self.sort: list[tuple[str, str]] = []
def sort_reversed(self) -> bool:
# Descending is the default
if self.sort:
# sort_reversed is True if we want to sort in ascending order
return self.sort[-1][-1] == "asc"
return False
def parse_query(query: str) -> ParsedQuery:
tokenizer = GenericTokenizer(query)
state = "COMMAND"
characters = ""
parsed_query = ParsedQuery()
for char in tokenizer:
if char.isspace():
if state == "SORT":
parsed_query.sort.append((characters, "desc"))
characters = ""
state = "SORT_ORDER"
if state == "COMMAND":
if characters.lower() in ["fields", "limit", "sort"]:
state = characters.upper()
else:
# Unknown/Unsupported command
pass
characters = ""
tokenizer.skip_white_space()
continue
if char == "|":
if state == "FIELDS":
parsed_query.fields.append(characters)
characters = ""
if state == "LIMIT":
parsed_query.limit = int(characters)
characters = ""
if state == "SORT_ORDER":
if characters != "":
parsed_query.sort[-1] = (parsed_query.sort[-1][0], characters)
characters = ""
state = "COMMAND"
tokenizer.skip_white_space()
continue
if char == ",":
if state == "FIELDS":
parsed_query.fields.append(characters)
characters = ""
continue
characters += char
if state == "FIELDS":
parsed_query.fields.append(characters)
if state == "LIMIT":
parsed_query.limit = int(characters)
if state == "SORT":
parsed_query.sort.append((characters, "desc"))
if state == "SORT_ORDER":
parsed_query.sort[-1] = (parsed_query.sort[-1][0], characters)
return parsed_query
|
