Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _add_operator_if_needed(self):
self.assert_no_raw_query()
if len(self._where_tokens) > 0:
query_operator = None
last_token = self._where_tokens[-1]
if last_token is not None and last_token.token in Query.where_operators:
query_operator = QueryOperator.AND if self.default_operator == QueryOperator.AND else QueryOperator.OR
search_operator = getattr(last_token, "search_operator", None)
if search_operator and search_operator != QueryOperator.OR:
# default to OR operator after search if AND was not specified explicitly
query_operator = QueryOperator.OR
if query_operator:
self._where_tokens.append(_Token(write=str(query_operator)))
def order_by(self, field_name, ordering=OrderingType.str, descending=False):
self.assert_no_raw_query()
field_name = Query.escape_if_needed(field_name)
self._order_by_tokens.append(
_Token(field_name=field_name, token="order_by", write=field_name, descending=descending,
ordering=ordering))
return self
def where_greater_than(self, field_name, value):
field_name = Query.escape_if_needed(field_name)
self._add_operator_if_needed()
self.negate_if_needed(field_name)
if isinstance(value, timedelta):
value = Utils.timedelta_tick(value)
token = _Token(field_name=field_name, token="greater_than", value=self.add_query_parameter(value))
token.write = self.rql_where_write(token)
self._where_tokens.append(token)
return self
def where_exists(self, field_name):
field_name = Query.escape_if_needed(field_name)
self._add_operator_if_needed()
self.negate_if_needed(field_name)
self._where_tokens.append(_Token(field_name=field_name, value=None, token="exists", write=")"))
return self
def where_between(self, field_name, start, end, exact=False):
field_name = Query.escape_if_needed(field_name)
self._add_operator_if_needed()
self.negate_if_needed(field_name)
if isinstance(start, timedelta):
start = Utils.timedelta_tick(start)
if isinstance(end, timedelta):
end = Utils.timedelta_tick(end)
from_parameter_name = self.add_query_parameter("*" if start is None else start)
to_parameter_name = self.add_query_parameter("NULL" if end is None else end)
token = _Token(field_name=field_name, token="between", value=(from_parameter_name, to_parameter_name),
exact=exact)
token.write = self.rql_where_write(token)
self._where_tokens.append(token)
return self