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