"""
DomainES
--------
.. code-block:: python
from corehq.apps.es import DomainES
query = (DomainES()
.in_domains(domains)
.created(gte=datespan.startdate, lte=datespan.enddate)
.size(0))
"""
from django_countries import Countries
from . import filters
from .client import ElasticDocumentAdapter, create_document_adapter
from .es_query import HQESQuery
from .index.analysis import COMMA_ANALYSIS
from .index.settings import IndexSettingsKey
HQ_DOMAINS_INDEX_CANONICAL_NAME = 'domains'
[docs]class DomainES(HQESQuery):
index = HQ_DOMAINS_INDEX_CANONICAL_NAME
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]class ElasticDomain(ElasticDocumentAdapter):
analysis = COMMA_ANALYSIS
settings_key = IndexSettingsKey.DOMAINS
canonical_name = HQ_DOMAINS_INDEX_CANONICAL_NAME
@property
def mapping(self):
from .mappings.domain_mapping import DOMAIN_MAPPING
return DOMAIN_MAPPING
@property
def model_cls(self):
from corehq.apps.domain.models import Domain
return Domain
def _from_dict(self, domain_dict):
"""
Takes a domain dict and applies required transformation to make it suitable for ES.
:param domain: an instance of ``dict`` which is result of ``Domain.to_json()``
"""
from corehq.apps.accounting.models import Subscription
sub = Subscription.visible_objects.filter(subscriber__domain=domain_dict['name'], is_active=True)
domain_dict['deployment'] = domain_dict.get('deployment') or {}
countries = domain_dict['deployment'].get('countries', [])
domain_dict['deployment']['countries'] = []
if sub:
domain_dict['subscription'] = sub[0].plan_version.plan.edition
for country in countries:
domain_dict['deployment']['countries'].append(Countries[country].upper())
return super()._from_dict(domain_dict)
domain_adapter = create_document_adapter(
ElasticDomain,
"hqdomains_2021-03-08",
"hqdomain",
)
[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")