Source code for pysnow.query_builder

# -*- coding: utf-8 -*-

import inspect
import six
import pytz

from .exceptions import (QueryEmpty,
                         QueryExpressionError,
                         QueryMissingField,
                         QueryMultipleExpressions,
                         QueryTypeError)


[docs]class QueryBuilder(object): """Query builder - for constructing advanced ServiceNow queries""" def __init__(self): self._query = [] self.current_field = None self.c_oper = None self.l_oper = None
[docs] def field(self, field): """Sets the field to operate on :param field: field (str) to operate on """ self.current_field = field return self
[docs] def order_descending(self): """Sets ordering of field descending""" self._query.append('ORDERBYDESC{0}'.format(self.current_field)) self.c_oper = inspect.currentframe().f_back.f_code.co_name return self
[docs] def order_ascending(self): """Sets ordering of field ascending""" self._query.append('ORDERBY{0}'.format(self.current_field)) self.c_oper = inspect.currentframe().f_back.f_code.co_name return self
[docs] def starts_with(self, starts_with): """Adds new `STARTSWITH` condition :param starts_with: Match field starting with the provided value """ return self._add_condition('STARTSWITH', starts_with, types=[str])
[docs] def ends_with(self, ends_with): """Adds new `ENDSWITH` condition :param ends_with: Match field ending with the provided value """ return self._add_condition('ENDSWITH', ends_with, types=[str])
[docs] def contains(self, contains): """Adds new `LIKE` condition :param contains: Match field containing the provided value """ return self._add_condition('LIKE', contains, types=[str])
[docs] def not_contains(self, not_contains): """Adds new `NOTLIKE` condition :param not_contains: Match field not containing the provided value """ return self._add_condition('NOTLIKE', not_contains, types=[str])
[docs] def is_empty(self): """Adds new `ISEMPTY` condition""" return self._add_condition('ISEMPTY', '', types=[str, int])
[docs] def is_not_empty(self): """Adds new `ISNOTEMPTY` condition""" return self._add_condition('ISNOTEMPTY', '', types=[str, int])
[docs] def equals(self, data): """Adds new `IN` or `=` condition depending on if a list or string was provided :param data: string or list of values :raise: - QueryTypeError: if `data` is of an unexpected type """ if isinstance(data, six.string_types): return self._add_condition('=', data, types=[int, str]) elif isinstance(data, list): return self._add_condition('IN', ",".join(map(str, data)), types=[str]) raise QueryTypeError('Expected value of type `str` or `list`, not %s' % type(data))
[docs] def not_equals(self, data): """Adds new `NOT IN` or `!=` condition depending on if a list or string was provided :param data: string or list of values :raise: - QueryTypeError: if `data` is of an unexpected type """ if isinstance(data, six.string_types): return self._add_condition('!=', data, types=[int, str]) elif isinstance(data, list): return self._add_condition('NOT IN', ",".join(data), types=[str]) raise QueryTypeError('Expected value of type `str` or `list`, not %s' % type(data))
[docs] def greater_than(self, greater_than): """Adds new `>` condition :param greater_than: str or datetime compatible object (naive UTC datetime or tz-aware datetime) :raise: - QueryTypeError: if `greater_than` is of an unexpected type """ if hasattr(greater_than, 'strftime'): greater_than = datetime_as_utc(greater_than).strftime('%Y-%m-%d %H:%M:%S') elif isinstance(greater_than, six.string_types): raise QueryTypeError('Expected value of type `int` or instance of `datetime`, not %s' % type(greater_than)) return self._add_condition('>', greater_than, types=[int, str])
[docs] def less_than(self, less_than): """Adds new `<` condition :param less_than: str or datetime compatible object (naive UTC datetime or tz-aware datetime) :raise: - QueryTypeError: if `less_than` is of an unexpected type """ if hasattr(less_than, 'strftime'): less_than = datetime_as_utc(less_than).strftime('%Y-%m-%d %H:%M:%S') elif isinstance(less_than, six.string_types): raise QueryTypeError('Expected value of type `int` or instance of `datetime`, not %s' % type(less_than)) return self._add_condition('<', less_than, types=[int, str])
[docs] def between(self, start, end): """Adds new `BETWEEN` condition :param start: int or datetime compatible object (in SNOW user's timezone) :param end: int or datetime compatible object (in SNOW user's timezone) :raise: - QueryTypeError: if start or end arguments is of an invalid type """ if hasattr(start, 'strftime') and hasattr(end, 'strftime'): dt_between = ( 'javascript:gs.dateGenerate("%(start)s")' "@" 'javascript:gs.dateGenerate("%(end)s")' ) % { 'start': start.strftime('%Y-%m-%d %H:%M:%S'), 'end': end.strftime('%Y-%m-%d %H:%M:%S') } elif isinstance(start, int) and isinstance(end, int): dt_between = '%d@%d' % (start, end) else: raise QueryTypeError("Expected `start` and `end` of type `int` " "or instance of `datetime`, not %s and %s" % (type(start), type(end))) return self._add_condition('BETWEEN', dt_between, types=[str])
[docs] def AND(self): """Adds an and-operator""" return self._add_logical_operator('^')
[docs] def OR(self): """Adds an or-operator""" return self._add_logical_operator('^OR')
[docs] def NQ(self): """Adds a NQ-operator (new query)""" return self._add_logical_operator('^NQ')
def _add_condition(self, operator, operand, types): """Appends condition to self._query after performing validation :param operator: operator (str) :param operand: operand :param types: allowed types :raise: - QueryMissingField: if a field hasn't been set - QueryMultipleExpressions: if a condition already has been set - QueryTypeError: if the value is of an unexpected type """ if not self.current_field: raise QueryMissingField("Conditions requires a field()") elif not type(operand) in types: caller = inspect.currentframe().f_back.f_code.co_name raise QueryTypeError("Invalid type passed to %s() , expected: %s" % (caller, types)) elif self.c_oper: raise QueryMultipleExpressions("Expected logical operator after expression") self.c_oper = inspect.currentframe().f_back.f_code.co_name self._query.append("%(current_field)s%(operator)s%(operand)s" % { 'current_field': self.current_field, 'operator': operator, 'operand': operand }) return self def _add_logical_operator(self, operator): """Adds a logical operator in query :param operator: logical operator (str) :raise: - QueryExpressionError: if a expression hasn't been set """ if not self.c_oper: raise QueryExpressionError("Logical operators must be preceded by an expression") self.current_field = None self.c_oper = None self.l_oper = inspect.currentframe().f_back.f_code.co_name self._query.append(operator) return self def __str__(self): """String representation of the query object :raise: - QueryEmpty: if there's no conditions defined - QueryMissingField: if field() hasn't been set - QueryExpressionError: if a expression hasn't been set :return: - String-type query """ if len(self._query) == 0: raise QueryEmpty("At least one condition is required") elif self.current_field is None: raise QueryMissingField("Logical operator expects a field()") elif self.c_oper is None: raise QueryExpressionError("field() expects an expression") return str().join(self._query)
def datetime_as_utc(date_obj): if date_obj.tzinfo is not None and date_obj.tzinfo.utcoffset(date_obj) is not None: return date_obj.astimezone(pytz.UTC) return date_obj