Source code for corehq.apps.es.domains

"""
DomainES
--------

Here's an example generating a histogram of domain creations (that's a type of
faceted query), filtered by a provided list of domains and a report date range.

.. code-block:: python

    from corehq.apps.es import DomainES

    domains_after_date = (DomainES()
                          .in_domains(domains)
                          .created(gte=datespan.startdate, lte=datespan.enddate)
                          .date_histogram('date', 'date_created', interval)
                          .size(0))
    histo_data = domains_after_date.run().aggregations.date.buckets_list
"""
from . import filters
from .es_query import HQESQuery


[docs]class DomainES(HQESQuery): index = 'domains' default_filters = { 'not_snapshot': filters.NOT(filters.term('is_snapshot', True)), }
[docs] def only_snapshots(self): """Normally snapshots are excluded, instead, return only snapshots""" return (self.remove_default_filter('not_snapshot') .filter(filters.term('is_snapshot', True)))
@property def builtin_filters(self): return [ non_test_domains, incomplete_domains, real_domains, created, last_modified, in_domains, is_active, is_active_project, created_by_user, self_started, ] + super(DomainES, self).builtin_filters
[docs]def non_test_domains(): return filters.term("is_test", [False, "none"])
[docs]def incomplete_domains(): return filters.OR(filters.missing("countries"), filters.missing("internal.area"), filters.missing("internal.notes"), filters.missing("internal.organization_name"), filters.missing("internal.sub_area"), )
[docs]def real_domains(): return filters.term("is_test", False)
[docs]def created(gt=None, gte=None, lt=None, lte=None): return filters.date_range('date_created', gt, gte, lt, lte)
[docs]def last_modified(gt=None, gte=None, lt=None, lte=None): return filters.date_range('last_modified', gt, gte, lt, lte)
[docs]def in_domains(domains): return filters.term('name', domains)
[docs]def is_active(is_active=True): return filters.term('is_active', is_active)
[docs]def is_active_project(is_active=True): # Project is active - has submitted a form in the last 30 days return filters.term('cp_is_active', is_active)
[docs]def created_by_user(creating_user): return filters.term('creating_user', creating_user)
[docs]def self_started(): return filters.term("internal.self_started", "true")