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