Source code for corehq.apps.es.sms

"""
SMSES
--------
"""
from . import filters
from .es_query import HQESQuery


[docs]class SMSES(HQESQuery): index = 'sms' @property def builtin_filters(self): return [ incoming_messages, outgoing_messages, to_commcare_user, to_commcare_case, to_web_user, to_couch_user, to_commcare_user_or_case, received, direction, processed, processed_or_incoming_messages, ] + super(SMSES, self).builtin_filters
[docs] def user_aggregation(self): return self.terms_aggregation('couch_recipient', 'user')
[docs]def incoming_messages(): return direction("i")
[docs]def outgoing_messages(): return direction("o")
[docs]def direction(direction_): return filters.term("direction", direction_)
[docs]def processed_or_incoming_messages(): return filters.NOT(filters.AND(outgoing_messages(), processed(False)))
[docs]def processed(processed=True): return filters.term('processed', processed)
[docs]def to_commcare_user(): return filters.term("couch_recipient_doc_type", "commcareuser")
[docs]def to_commcare_case(): return filters.term("couch_recipient_doc_type", "commcarecase")
[docs]def to_web_user(): return filters.term("couch_recipient_doc_type", "webuser")
[docs]def to_couch_user(): return filters.term("couch_recipient_doc_type", "couchuser")
[docs]def to_commcare_user_or_case(): return filters.OR(to_commcare_user(), to_commcare_case())
[docs]def received(gt=None, gte=None, lt=None, lte=None): return filters.date_range('date', gt, gte, lt, lte)