"""
The Simple Query Language provides a simplified syntax to retrieve entities
which match a set of conditions. A query is composed by a list of
statements separated by the ';' character. Each statement expresses a
matching condition. The query returns all the entities that match all
the matching conditions (AND logical operator).
For further details of the language please refer to:
https://telefonicaid.github.io/fiware-orion/api/v2/stable/
"""
import regex as re
from aenum import Enum
from typing import Union, List, Tuple, Any
[docs]class Operator(str, Enum):
"""
The list of operators (and the format of the values they use) is as follows:
"""
_init_ = "value __doc__"
EQUAL = (
"==",
"Single element, e.g. temperature!=41. For an entity to "
"match, it must contain the target property (temperature) "
"and the target property value must not be the query value "
"(41). "
"A list of comma-separated values, e.g. color!=black,"
"red. For an entity to match, it must contain the target "
"property and the target property value must not be any "
"of the values in the list (AND clause) (or not include any "
"of the values in the list in case the target property "
"value is an array). Eg. entities whose attribute color is "
"set to black will not match, while entities whose "
"attribute color is set to white will match."
"A range, specified as a minimum and maximum separated by "
".., e.g. temperature!=10..20. For an entity to match, "
"it must contain the target property (temperature) and the "
"target property value must not be between the upper and "
"lower limits (both included). Ranges can only be used "
"with elements target properties that represent dates (in "
"ISO8601 format), numbers or strings. ",
)
UNEQUAL = (
"!=",
"Single element, e.g. temperature!=41. For an entity to "
"match, it must contain the target property "
"(temperature) and the target property value must not be "
"the query value (41). A list of comma-separated values, "
"e.g. color!=black,red. For an entity to match, it must "
"contain the target property and the target property "
"value must not be any of the values in the list (AND "
"clause) (or not include any of the values in the list "
"in case the target property value is an array). Eg. "
"entities whose attribute color is set to black will not "
"match, while entities whose attribute color is set to "
"white will match. A range, specified as a minimum and "
"maximum separated by .., e.g. temperature!=10..20. For "
"an entity to match, it must contain the target property "
"(temperature) and the target property value must not be "
"between the upper and lower limits (both included). "
"Ranges can only be used with elements target properties "
"that represent dates (in ISO8601 format), numbers or "
"strings. ",
)
GREATER_THAN = (
">",
"The right-hand side must be a single element, e.g. "
"temperature>42. For an entity to match, it must "
"contain the target property (temperature) and the "
"target property value must be strictly greater than "
"the query value (42). This operation is only valid "
"for target properties of type date, number or "
"string (used with target properties of other types "
"may lead to unpredictable results). ",
)
LESS_THAN = (
"<",
"The right-hand side must be a single element, e.g. "
"temperature<43. For an entity to match, it must "
"contain the target property (temperature) and the "
"target property value must be strictly less than the "
"value (43). This operation is only valid for target "
"properties of type date, number or string (used with "
"target properties of other types may lead to "
"unpredictable results). ",
)
GREATER_OR_EQUAL = (
">=",
"The right-hand side must be a single element, "
"e.g. temperature>=44. For an entity to match, "
"it must contain the target property ("
"temperature) and the target property value "
"must be greater than or equal to that value "
"(44). This operation is only valid for target "
"properties of type date, number or string "
"(used with target properties of other types "
"may lead to unpredictable results). ",
)
LESS_OR_EQUAL = (
"<=",
"The right-hand side must be a single element, "
"e.g. temperature<=45. For an entity to match, "
"it must contain the target property (temperature) "
"and the target property value must be less than "
"or equal to that value (45). This operation is "
"only valid for target properties of type date, "
"number or string (used with target properties of "
"other types may lead to unpredictable results). ",
)
MATCH_PATTERN = (
"~=",
"The value matches a given pattern, expressed as a "
"regular expression, e.g. color~=ow. For an entity "
"to match, it must contain the target property ("
"color) and the target property value must match "
"the string in the right-hand side, 'ow' in this "
"example (brown and yellow would match, black and "
"white would not). This operation is only valid "
"for target properties of type string. ",
)
[docs] @classmethod
def list(cls):
"""
Returns:
list of all valid values
"""
return list(map(lambda c: c.value, cls))
[docs]class QueryStatement(Tuple):
"""
Simple query statement
"""
def __new__(cls, left: str, op: Union[str, Operator], right: Any):
q = tuple.__new__(QueryStatement, (left, op, right))
q = cls.validate(q)
return q
@classmethod
def __get_validators__(cls):
yield cls.validate
[docs] @classmethod
def validate(cls, value):
"""
Validates statements
Args:
value:
Returns:
"""
if isinstance(value, (tuple, QueryStatement)):
if len(value) != 3:
raise TypeError("3-tuple required")
if not isinstance(value[0], str):
raise TypeError("First argument must be a string!")
if value[1] not in Operator.list():
raise TypeError("Invalid comparison operator!")
if value[1] not in [
Operator.EQUAL,
Operator.UNEQUAL,
Operator.MATCH_PATTERN,
]:
try:
float(value[2])
except ValueError as err:
err.args += (
"Invalid combination of operator and right " "hand side!",
)
raise
return value
elif isinstance(value, str):
return cls.parse_str(value)
raise TypeError
[docs] def to_str(self):
"""
Parses QueryStatement to String
Returns:
String
"""
if not isinstance(self[2], str):
right = str(self[2])
elif self[2].isnumeric():
right = f"{self[2]}"
else:
right = self[2]
return "".join([self[0], self[1], right])
[docs] @classmethod
def parse_str(cls, string: str):
"""
Generates QueryStatement form string
Args:
string:
Returns:
QueryStatement
"""
for op in Operator.list():
if re.fullmatch(rf"^\w((\w|[^&,?,/,#,\*,\s]\w)?)*{op}\w+$", string):
args = string.split(op)
if len(args) == 2:
if args[1].isnumeric():
try:
right = int(args[1])
except ValueError:
right = float(args[1])
return QueryStatement(args[0], op, right)
return QueryStatement(args[0], op, args[1])
raise ValueError
def __str__(self):
"""Return str(self)."""
return self.to_str()
def __repr__(self):
"""Return repr(self)."""
return self.to_str().__repr__()
[docs]class QueryString:
"""
Class for validated QueryStrings that can be used in api clients
"""
def __init__(
self, qs: Union[Tuple, QueryStatement, List[Union[QueryStatement, Tuple]]]
):
qs = self.__check_arguments(qs=qs)
self._qs = qs
@classmethod
def __check_arguments(cls, qs):
"""
Check arguments on consistency
Args:
qs: queny statement object
returns:
List of QueryStatements
"""
if isinstance(qs, List):
for idx, item in enumerate(qs):
if not isinstance(item, QueryStatement):
qs[idx] = QueryStatement(*item)
# Remove duplicates
qs = list(dict.fromkeys(qs))
elif isinstance(qs, QueryStatement):
qs = [qs]
elif isinstance(qs, tuple):
qs = [QueryStatement(*qs)]
else:
raise ValueError("Invalid argument!")
return qs
[docs] def update(self, qs: Union[Tuple, QueryStatement, List[QueryStatement]]):
"""
Adds or updates QueryStatement within QueryString. First to arguments
must match an existing argument for update. This redundant rules
Args:
qs: Query statement to add to the string object
Returns:
None
"""
qs = self.__check_arguments(qs=qs)
self._qs.extend(qs)
self._qs = list(dict.fromkeys(qs))
[docs] def remove(self, qs: Union[Tuple, QueryStatement, List[QueryStatement]]):
"""
Remove Statement from QueryString
Args:
qs:
Returns:
"""
qs = self.__check_arguments(qs=qs)
for q in qs:
self._qs.remove(q)
@classmethod
def __get_validators__(cls):
yield cls.validate
[docs] @classmethod
def validate(cls, v):
"""validate QueryString"""
if isinstance(v, QueryString):
return v
if isinstance(v, str):
return cls.parse_str(v)
raise ValueError("Invalid argument!")
[docs] def to_str(self):
"""
Parsing self.qs to string object
Returns:
String: query string that can be added to requests as parameter
"""
return ";".join([q.to_str() for q in self._qs])
[docs] @classmethod
def parse_str(cls, string: str):
"""
Creates QueryString from string
Args:
string:
Returns:
QueryString
"""
q_parts = string.split(";")
qs = []
for part in q_parts:
q = QueryStatement.parse_str(part)
qs.append(q)
return QueryString(qs=qs)
def __str__(self):
"""Return str(self)."""
return self.to_str()
def __repr__(self):
"""Return repr(self)."""
return self.to_str().__repr__()