[fix bug 907933] Changes in the codebase for the ElasticSearch upgrade.

.gitmodules

@@ -11,9 +11,6 @@
 [submodule "vendor-local/src/elasticutils"]
 	path = vendor-local/src/elasticutils
 	url = git://github.com/mozilla/elasticutils
-[submodule "vendor-local/src/pyes"]
-	path = vendor-local/src/pyes
-	url = git://github.com/aparo/pyes
 [submodule "vendor-local/src/django-tastypie"]
 	path = vendor-local/src/django-tastypie
 	url = git://github.com/toastdriven/django-tastypie

docs/installation/installation-virtualenv.rst

@@ -102,12 +102,14 @@ When you want to start contributing...
 
 #. Download ElasticSearch::
 
-     (venv)$ wget https://download.elasticsearch.org/elasticsearch/elasticsearch/elasticsearch-0.90.10.tar.gz
-     (venv)$ tar zxf elasticsearch-0.90.10.tar.gz
+     (venv)$ wget https://download.elasticsearch.org/elasticsearch/elasticsearch/elasticsearch-1.2.4.tar.gz
+     (venv)$ tar zxf elasticsearch-1.2.4.tar.gz
 
    and run::
 
-     (venv)$ ./elasticsearch-0.90.10/bin/elasticsearch
+     (venv)$ ./elasticsearch-1.2.4/bin/elasticsearch -d
+
+  This will run the elasticsearch instance in the background.
 
 #. Update product details::
 

mozillians/common/tests/__init__.py

@@ -24,6 +24,14 @@ ES_INDEXES = {
 @override_settings(AUTHENTICATION_BACKENDS=AUTHENTICATION_BACKENDS,
                    ES_INDEXES=ES_INDEXES)
 class TestCase(BaseTestCase):
+    def __init__(self, *args, **kwargs):
+        from elasticutils.contrib.django import get_es
+        es = get_es()
+
+        es.indices.create(index=ES_INDEXES['default'], ignore=400)
+        es.indices.create(index=ES_INDEXES['public'], ignore=400)
+        super(TestCase, self).__init__(*args, **kwargs)
+
     @contextmanager
     def login(self, user):
         client = Client()

mozillians/phonebook/tests/test_views/test_views_delete.py

@@ -7,7 +7,7 @@ from mock import patch, call
 from nose.tools import eq_, ok_
 
 from mozillians.common.tests import TestCase, requires_login
-from mozillians.users.models import UserProfile
+from mozillians.users.es import UserProfileMappingType
 from mozillians.users.tests import UserFactory
 
 
@@ -61,9 +61,8 @@ class DeleteTests(TestCase):
         remove_from_basket_task_mock.assert_called_with(
             user.email, user.userprofile.basket_token)
         unindex_objects_mock.assert_has_calls([
-            call(UserProfile, [user.userprofile.id], public_index=False),
-            call(UserProfile, [user.userprofile.id], public_index=True)
-            ])
+            call(UserProfileMappingType, [user.userprofile.id], public_index=False),
+            call(UserProfileMappingType, [user.userprofile.id], public_index=True)])
         ok_(not User.objects.filter(username=user.username).exists())
 
     @patch('mozillians.users.models.remove_from_basket_task.delay')
@@ -81,7 +80,6 @@ class DeleteTests(TestCase):
         remove_from_basket_task_mock.assert_called_with(
             user.email, user.userprofile.basket_token)
         unindex_objects_mock.assert_has_calls([
-            call(UserProfile, [user.userprofile.id], public_index=False),
-            call(UserProfile, [user.userprofile.id], public_index=True)
-            ])
+            call(UserProfileMappingType, [user.userprofile.id], public_index=False),
+            call(UserProfileMappingType, [user.userprofile.id], public_index=True)])
         ok_(not User.objects.filter(username=user.username).exists())

mozillians/phonebook/views.py

@@ -23,7 +23,7 @@ from mozillians.groups.models import Group
 from mozillians.phonebook.models import Invite
 from mozillians.phonebook.utils import redeem_invite
 from mozillians.users.managers import EMPLOYEES, MOZILLIANS, PUBLIC, PRIVILEGED
-from mozillians.users.models import UserProfile
+from mozillians.users.models import UserProfile, UserProfileMappingType
 
 
 @allow_unvouched
@@ -267,8 +267,8 @@ def search(request):
         public = not (request.user.is_authenticated()
                       and request.user.userprofile.is_vouched)
 
-        profiles = UserProfile.search(query, public=public,
-                                      include_non_vouched=include_non_vouched)
+        profiles = UserProfileMappingType.search(
+            query, public=public, include_non_vouched=include_non_vouched)
         if not public:
             groups = Group.search(query)
 
@@ -323,7 +323,9 @@ def betasearch(request):
                       and request.user.userprofile.is_vouched)
 
         profiles_matching_filter = list(filtr.qs.values_list('id', flat=True))
-        profiles = UserProfile.search(query, include_non_vouched=True, public=public)
+        profiles = UserProfileMappingType.search(query,
+                                                 include_non_vouched=True,
+                                                 public=public)
         profiles = profiles.filter(id__in=profiles_matching_filter)
 
         paginator = Paginator(profiles, limit)

mozillians/settings/base.py

@@ -228,7 +228,7 @@ CSP_STYLE_SRC = ("'self'",
 
 # Elasticutils settings
 ES_DISABLED = True
-ES_HOSTS = ['127.0.0.1:9200']
+ES_URLS = ['http://127.0.0.1:9200']
 ES_INDEXES = {'default': 'mozillians',
               'public': 'mozillians-public'}
 ES_INDEXING_TIMEOUT = 10

mozillians/users/cron.py

@@ -1,14 +1,12 @@
 from django.conf import settings
 
 import cronjobs
-import pyes.exceptions
 
 from celery.task.sets import TaskSet
-from celeryutils import chunked
 from elasticutils.contrib.django import get_es
 
 from mozillians.users.tasks import index_objects
-from mozillians.users.models import PUBLIC, UserProfile
+from mozillians.users.models import PUBLIC, UserProfile, UserProfileMappingType
 
 
 @cronjobs.register
@@ -16,26 +14,29 @@ def index_all_profiles():
     # Get an es object, delete index and re-create it
     es = get_es(timeout=settings.ES_INDEXING_TIMEOUT)
     mappings = {'mappings':
-                {UserProfile._meta.db_table: UserProfile.get_mapping()}}
+                {UserProfileMappingType.get_mapping_type_name():
+                 UserProfileMappingType.get_mapping()}}
 
     def _recreate_index(index):
-        try:
-            es.delete_index_if_exists(index)
-        except pyes.exceptions.IndexMissingException:
-            pass
-        es.create_index(index, settings=mappings)
+        es.indices.delete(index=index, ignore=[400, 404])
+        es.indices.create(index, body=mappings)
+
     _recreate_index(settings.ES_INDEXES['default'])
     _recreate_index(settings.ES_INDEXES['public'])
 
     # mozillians index
     ids = UserProfile.objects.complete().values_list('id', flat=True)
-    ts = [index_objects.subtask(args=[UserProfile, chunk, False])
-          for chunk in chunked(sorted(list(ids)), 150)]
+    ts = [index_objects.subtask(kwargs={'mapping_type': UserProfileMappingType,
+                                        'ids': ids,
+                                        'chunk_size': 150,
+                                        'public_index': False})]
 
     # public index
     ids = (UserProfile.objects.complete().public_indexable()
            .privacy_level(PUBLIC).values_list('id', flat=True))
-    ts += [index_objects.subtask(args=[UserProfile, chunk, True])
-           for chunk in chunked(sorted(list(ids)), 150)]
+    ts += [index_objects.subtask(kwargs={'mapping_type': UserProfileMappingType,
+                                         'ids': ids,
+                                         'chunk_size': 150,
+                                         'public_index': True})]
 
     TaskSet(ts).apply_async()

mozillians/users/es.py

@@ -0,0 +1,203 @@
+from django.conf import settings
+
+from elasticsearch import TransportError
+from elasticsearch.exceptions import NotFoundError
+from elasticutils.contrib.django import Indexable, MappingType, S, get_es
+
+from mozillians.phonebook.helpers import langcode_to_name
+from mozillians.users.managers import MOZILLIANS, PUBLIC
+
+ES_MAPPING_TYPE_NAME = 'user-profile'
+
+
+class PrivacyAwareS(S):
+
+    def privacy_level(self, level=MOZILLIANS):
+        """Set privacy level for query set."""
+        self._privacy_level = level
+        return self
+
+    def _clone(self, *args, **kwargs):
+        new = super(PrivacyAwareS, self)._clone(*args, **kwargs)
+        new._privacy_level = getattr(self, '_privacy_level', None)
+        return new
+
+    def __iter__(self):
+        self._iterator = super(PrivacyAwareS, self).__iter__()
+
+        def _generator():
+            while True:
+                obj = self._iterator.next()
+                obj._privacy_level = getattr(self, '_privacy_level', None)
+                yield obj.get_object()
+        return _generator()
+
+
+class UserProfileMappingType(MappingType, Indexable):
+
+    @classmethod
+    def get_index(cls, public_index=False):
+        if public_index:
+            return settings.ES_INDEXES['public']
+        return settings.ES_INDEXES['default']
+
+    @classmethod
+    def get_mapping_type_name(cls):
+        return ES_MAPPING_TYPE_NAME
+
+    @classmethod
+    def get_model(cls):
+        from mozillians.users.models import UserProfile
+        return UserProfile
+
+    @classmethod
+    def get_es(cls):
+        return get_es(urls=settings.ES_URLS)
+
+    @classmethod
+    def get_mapping(cls):
+        """Returns an ElasticSearch mapping."""
+        return {
+            'properties': {
+                'id': {'type': 'integer'},
+                'name': {'type': 'string', 'index': 'not_analyzed'},
+                'fullname': {'type': 'string', 'analyzer': 'standard'},
+                'email': {'type': 'string', 'index': 'not_analyzed'},
+                'ircname': {'type': 'string', 'index': 'not_analyzed'},
+                'username': {'type': 'string', 'index': 'not_analyzed'},
+                'country': {'type': 'string', 'analyzer': 'whitespace'},
+                'region': {'type': 'string', 'analyzer': 'whitespace'},
+                'city': {'type': 'string', 'analyzer': 'whitespace'},
+                'skills': {'type': 'string', 'analyzer': 'whitespace'},
+                'groups': {'type': 'string', 'analyzer': 'whitespace'},
+                'languages': {'type': 'string', 'index': 'not_analyzed'},
+                'bio': {'type': 'string', 'analyzer': 'snowball'},
+                'is_vouched': {'type': 'boolean'},
+                'allows_mozilla_sites': {'type': 'boolean'},
+                'allows_community_sites': {'type': 'boolean'},
+                'photo': {'type': 'boolean'},
+                'last_updated': {'type': 'date'},
+                'date_joined': {'type': 'date'}
+            }
+        }
+
+    @classmethod
+    def index(cls, document, id_=None, overwrite_existing=False, es=None,
+              public_index=False):
+        """ Overide elasticutils.index() to support more than one index
+        for UserProfile model.
+
+        """
+        if es is None:
+            es = get_es()
+
+        es.index(document, index=cls.get_index(public_index),
+                 doc_type=cls.get_mapping_type(),
+                 id=id_, overwrite_existing=overwrite_existing)
+
+    @classmethod
+    def refresh_index(cls, es=None, public_index=False):
+        if es is None:
+            es = get_es()
+        index = cls.get_index(public_index)
+        if es.indices.exists(index):
+            es.indices.refresh(index=index)
+
+    @classmethod
+    def unindex(cls, id_, es=None, public_index=False):
+        if es is None:
+            es = get_es()
+        try:
+            es.delete(index=cls.get_index(public_index),
+                      doc_type=cls.get_mapping_type_name(), id=id_)
+        except NotFoundError:
+            pass
+        except TransportError, e:
+            raise e
+
+    @classmethod
+    def extract_document(cls, obj_id, obj=None):
+        """Extract the following fields from a document."""
+
+        if obj is None:
+            obj = cls.get_model().objects.get(pk=obj_id)
+        doc = {}
+
+        attrs = ('id', 'is_vouched', 'ircname',
+                 'allows_mozilla_sites', 'allows_community_sites')
+        for a in attrs:
+            data = getattr(obj, a)
+            if isinstance(data, basestring):
+                data = data.lower()
+            doc.update({a: data})
+
+        doc['country'] = ([obj.geo_country.name, obj.geo_country.code]
+                          if obj.geo_country else None)
+        doc['region'] = obj.geo_region.name if obj.geo_region else None
+        doc['city'] = obj.geo_city.name if obj.geo_city else None
+
+        # user data
+        attrs = ('username', 'email', 'last_login', 'date_joined')
+        for a in attrs:
+            data = getattr(obj.user, a)
+            if isinstance(data, basestring):
+                data = data.lower()
+            doc.update({a: data})
+
+        doc.update(dict(fullname=obj.full_name.lower()))
+        doc.update(dict(name=obj.full_name.lower()))
+        doc.update(dict(bio=obj.bio))
+        doc.update(dict(has_photo=bool(obj.photo)))
+
+        for attribute in ['groups', 'skills']:
+            groups = []
+            for g in getattr(obj, attribute).all():
+                groups.extend(g.aliases.values_list('name', flat=True))
+            doc[attribute] = groups
+        # Add to search index language code, language name in English
+        # native lanugage name.
+        languages = []
+        for code in obj.languages.values_list('code', flat=True):
+            languages.append(code)
+            languages.append(langcode_to_name(code, 'en_US').lower())
+            languages.append(langcode_to_name(code, code).lower())
+        doc['languages'] = list(set(languages))
+        return doc
+
+    @classmethod
+    def get_indexable(cls):
+        model = cls.get_model()
+        return model.objects.order_by('id').values_list('id', flat=True)
+
+    @classmethod
+    def search(cls, query, include_non_vouched=False, public=False):
+        """Sensible default search for UserProfiles."""
+        query = query.lower().strip()
+        fields = ('username', 'bio__match', 'email', 'ircname',
+                  'country__match', 'country__match_phrase',
+                  'region__match', 'region__match_phrase',
+                  'city__match', 'city__match_phrase',
+                  'fullname__match', 'fullname__match_phrase',
+                  'fullname__prefix', 'fullname__fuzzy'
+                  'groups__match')
+        search = PrivacyAwareS(cls)
+        if public:
+            search = search.privacy_level(PUBLIC)
+        search = search.indexes(cls.get_index(public))
+
+        if query:
+            query_dict = dict((field, query) for field in fields)
+            search = (search.boost(fullname__match_phrase=5, username=5,
+                                   email=5, ircname=5, fullname__match=4,
+                                   country__match_phrase=4,
+                                   region__match_phrase=4,
+                                   city__match_phrase=4, fullname__prefix=3,
+                                   fullname__fuzzy=2, bio__match=2)
+                      .query(or_=query_dict))
+
+        search = search.order_by('_score', 'name')
+
+        if not include_non_vouched:
+            search = search.filter(is_vouched=True)
+
+        return search

mozillians/users/models.py

@@ -15,8 +15,6 @@ from django.template.loader import get_template
 
 
 import basket
-from elasticutils.contrib.django import S, get_es
-from elasticutils.contrib.django.models import SearchMixin
 from funfactory.urlresolvers import reverse
 from product_details import product_details
 from pytz import common_timezones
@@ -29,10 +27,10 @@ from mozillians.common.helpers import gravatar
 from mozillians.common.helpers import offset_of_timezone
 from mozillians.groups.models import (Group, GroupAlias, GroupMembership,
                                       Skill, SkillAlias)
-from mozillians.phonebook.helpers import langcode_to_name
 from mozillians.phonebook.validators import (validate_email, validate_twitter,
                                              validate_website, validate_username_not_url,
                                              validate_phone_number)
+from mozillians.users.es import UserProfileMappingType
 from mozillians.users import get_languages_for_locale
 from mozillians.users.managers import (EMPLOYEES,
                                        MOZILLIANS, PRIVACY_CHOICES, PRIVILEGED,
@@ -62,29 +60,6 @@ class PrivacyField(models.PositiveSmallIntegerField):
 add_introspection_rules([], ['^mozillians\.users\.models\.PrivacyField'])
 
 
-class PrivacyAwareS(S):
-
-    def privacy_level(self, level=MOZILLIANS):
-        """Set privacy level for query set."""
-        self._privacy_level = level
-        return self
-
-    def _clone(self, *args, **kwargs):
-        new = super(PrivacyAwareS, self)._clone(*args, **kwargs)
-        new._privacy_level = getattr(self, '_privacy_level', None)
-        return new
-
-    def __iter__(self):
-        self._iterator = super(PrivacyAwareS, self).__iter__()
-
-        def _generator():
-            while True:
-                obj = self._iterator.next()
-                obj._privacy_level = getattr(self, '_privacy_level', None)
-                yield obj
-        return _generator()
-
-
 class UserProfilePrivacyModel(models.Model):
     _privacy_level = None
 
@@ -144,14 +119,15 @@ class UserProfilePrivacyModel(models.Model):
                 else:
                     default = field.get_default()
                 privacy_fields[name] = default
-            # HACK: There's not really an email field on UserProfile, but it's faked with a property
+            # HACK: There's not really an email field on UserProfile,
+            # but it's faked with a property
             privacy_fields['email'] = u''
 
             cls.CACHED_PRIVACY_FIELDS = privacy_fields
         return cls.CACHED_PRIVACY_FIELDS
 
 
-class UserProfile(UserProfilePrivacyModel, SearchMixin):
+class UserProfile(UserProfilePrivacyModel):
     REFERRAL_SOURCE_CHOICES = (
         ('direct', 'Mozillians'),
         ('contribute', 'Get Involved'),
@@ -224,6 +200,13 @@ class UserProfile(UserProfilePrivacyModel, SearchMixin):
                                        choices=REFERRAL_SOURCE_CHOICES,
                                        default='direct')
 
+    def __unicode__(self):
+        """Return this user's name when their profile is called."""
+        return self.display_name
+
+    def get_absolute_url(self):
+        return reverse('phonebook:profile_view', args=[self.user.username])
+
     class Meta:
         db_table = 'profile'
         ordering = ['full_name']
@@ -285,109 +268,6 @@ class UserProfile(UserProfilePrivacyModel, SearchMixin):
     def _vouches_received(self):
         return self._vouches('vouches_received')
 
-    @classmethod
-    def extract_document(cls, obj_id, obj=None):
-        """Method used by elasticutils."""
-        if obj is None:
-            obj = cls.objects.get(pk=obj_id)
-        d = {}
-
-        attrs = ('id', 'is_vouched', 'ircname',
-                 'allows_mozilla_sites', 'allows_community_sites')
-        for a in attrs:
-            data = getattr(obj, a)
-            if isinstance(data, basestring):
-                data = data.lower()
-            d.update({a: data})
-
-        d['country'] = [obj.geo_country.name, obj.geo_country.code] if obj.geo_country else None
-        d['region'] = obj.geo_region.name if obj.geo_region else None
-        d['city'] = obj.geo_city.name if obj.geo_city else None
-
-        # user data
-        attrs = ('username', 'email', 'last_login', 'date_joined')
-        for a in attrs:
-            data = getattr(obj.user, a)
-            if isinstance(data, basestring):
-                data = data.lower()
-            d.update({a: data})
-
-        d.update(dict(fullname=obj.full_name.lower()))
-        d.update(dict(name=obj.full_name.lower()))
-        d.update(dict(bio=obj.bio))
-        d.update(dict(has_photo=bool(obj.photo)))
-
-        for attribute in ['groups', 'skills']:
-            groups = []
-            for g in getattr(obj, attribute).all():
-                groups.extend(g.aliases.values_list('name', flat=True))
-            d[attribute] = groups
-        # Add to search index language code, language name in English
-        # native lanugage name.
-        languages = []
-        for code in obj.languages.values_list('code', flat=True):
-            languages.append(code)
-            languages.append(langcode_to_name(code, 'en_US').lower())
-            languages.append(langcode_to_name(code, code).lower())
-        d['languages'] = list(set(languages))
-        return d
-
-    @classmethod
-    def get_mapping(cls):
-        """Returns an ElasticSearch mapping."""
-        return {
-            'properties': {
-                'id': {'type': 'integer'},
-                'name': {'type': 'string', 'index': 'not_analyzed'},
-                'fullname': {'type': 'string', 'analyzer': 'standard'},
-                'email': {'type': 'string', 'index': 'not_analyzed'},
-                'ircname': {'type': 'string', 'index': 'not_analyzed'},
-                'username': {'type': 'string', 'index': 'not_analyzed'},
-                'country': {'type': 'string', 'analyzer': 'whitespace'},
-                'region': {'type': 'string', 'analyzer': 'whitespace'},
-                'city': {'type': 'string', 'analyzer': 'whitespace'},
-                'skills': {'type': 'string', 'analyzer': 'whitespace'},
-                'groups': {'type': 'string', 'analyzer': 'whitespace'},
-                'languages': {'type': 'string', 'index': 'not_analyzed'},
-                'bio': {'type': 'string', 'analyzer': 'snowball'},
-                'is_vouched': {'type': 'boolean'},
-                'allows_mozilla_sites': {'type': 'boolean'},
-                'allows_community_sites': {'type': 'boolean'},
-                'photo': {'type': 'boolean'},
-                'last_updated': {'type': 'date'},
-                'date_joined': {'type': 'date'}}}
-
-    @classmethod
-    def search(cls, query, include_non_vouched=False, public=False):
-        """Sensible default search for UserProfiles."""
-        query = query.lower().strip()
-        fields = ('username', 'bio__text', 'email', 'ircname',
-                  'country__text', 'country__text_phrase',
-                  'region__text', 'region__text_phrase',
-                  'city__text', 'city__text_phrase',
-                  'fullname__text', 'fullname__text_phrase',
-                  'fullname__prefix', 'fullname__fuzzy'
-                  'groups__text')
-        s = PrivacyAwareS(cls)
-        if public:
-            s = s.privacy_level(PUBLIC)
-        s = s.indexes(cls.get_index(public))
-
-        if query:
-            q = dict((field, query) for field in fields)
-            s = (s.boost(fullname__text_phrase=5, username=5, email=5,
-                         ircname=5, fullname__text=4, country__text_phrase=4,
-                         region__text_phrase=4, city__text_phrase=4,
-                         fullname__prefix=3, fullname__fuzzy=2,
-                         bio__text=2).query(or_=q))
-
-        s = s.order_by('_score', 'name')
-
-        if not include_non_vouched:
-            s = s.filter(is_vouched=True)
-
-        return s
-
     @property
     def accounts(self):
         accounts_query = self.externalaccount_set.exclude(type=ExternalAccount.TYPE_WEBSITE)
@@ -490,13 +370,6 @@ class UserProfile(UserProfilePrivacyModel, SearchMixin):
             return vouches[0].date
         return None
 
-    def __unicode__(self):
-        """Return this user's name when their profile is called."""
-        return self.display_name
-
-    def get_absolute_url(self):
-        return reverse('phonebook:profile_view', args=[self.user.username])
-
     def set_instance_privacy_level(self, level):
         """Sets privacy level of instance."""
         self._privacy_level = level
@@ -700,43 +573,6 @@ class UserProfile(UserProfilePrivacyModel, SearchMixin):
         # create foreign keys without a database id.
         self.auto_vouch()
 
-    @classmethod
-    def get_index(cls, public_index=False):
-        if public_index:
-            return settings.ES_INDEXES['public']
-        return settings.ES_INDEXES['default']
-
-    @classmethod
-    def refresh_index(cls, timesleep=0, es=None, public_index=False):
-        if es is None:
-            es = get_es()
-
-        es.refresh(cls.get_index(public_index), timesleep=timesleep)
-
-    @classmethod
-    def index(cls, document, id_=None, bulk=False, force_insert=False,
-              es=None, public_index=False):
-        """ Overide elasticutils.index() to support more than one index
-        for UserProfile model.
-
-        """
-        if bulk and es is None:
-            raise ValueError('bulk is True, but es is None')
-
-        if es is None:
-            es = get_es()
-
-        es.index(document, index=cls.get_index(public_index),
-                 doc_type=cls.get_mapping_type(),
-                 id=id_, bulk=bulk, force_insert=force_insert)
-
-    @classmethod
-    def unindex(cls, id, es=None, public_index=False):
-        if es is None:
-            es = get_es()
-
-        es.delete(cls.get_index(public_index), cls.get_mapping_type(), id)
-
     def reverse_geocode(self):
         """
         Use the user's lat and lng to set their city, region, and country.
@@ -788,18 +624,18 @@ def update_basket(sender, instance, **kwargs):
           dispatch_uid='update_search_index_sig')
 def update_search_index(sender, instance, **kwargs):
     if instance.is_complete:
-        index_objects.delay(sender, [instance.id], public_index=False)
+        index_objects.delay(UserProfileMappingType, [instance.id], public_index=False)
         if instance.is_public_indexable:
-            index_objects.delay(sender, [instance.id], public_index=True)
+            index_objects.delay(UserProfileMappingType, [instance.id], public_index=True)
         else:
-            unindex_objects.delay(UserProfile, [instance.id], public_index=True)
+            unindex_objects.delay(UserProfileMappingType, [instance.id], public_index=True)
 
 
 @receiver(dbsignals.pre_delete, sender=UserProfile,
           dispatch_uid='remove_from_search_index_sig')
 def remove_from_search_index(sender, instance, **kwargs):
-    unindex_objects.delay(UserProfile, [instance.id], public_index=False)
-    unindex_objects.delay(UserProfile, [instance.id], public_index=True)
+    unindex_objects.delay(UserProfileMappingType, [instance.id], public_index=False)
+    unindex_objects.delay(UserProfileMappingType, [instance.id], public_index=True)
 
 
 @receiver(dbsignals.pre_delete, sender=UserProfile,

mozillians/users/tasks.py

@@ -8,10 +8,10 @@ from django.db.models import get_model
 
 import basket
 import requests
-import pyes
 from celery.task import task
 from celery.exceptions import MaxRetriesExceededError
 from elasticutils.contrib.django import get_es
+from elasticutils.utils import chunked
 
 from mozillians.users.managers import PUBLIC
 
@@ -219,41 +219,35 @@ def remove_from_basket_task(email, basket_token):
 
 
 @task
-def index_objects(model, ids, public_index, **kwargs):
+def index_objects(mapping_type, ids, chunk_size=100, public_index=False, **kwargs):
     if getattr(settings, 'ES_DISABLED', False):
         return
 
     es = get_es()
-    qs = model.objects.filter(id__in=ids)
-    if public_index:
-        qs = model.objects.privacy_level(PUBLIC).filter(id__in=ids)
+    model = mapping_type.get_model()
 
-    for item in qs:
-        model.index(model.extract_document(item.id, item),
-                    bulk=True, id_=item.id, es=es, public_index=public_index)
+    for id_list in chunked(ids, chunk_size):
+        documents = []
+        qs = model.objects.filter(id__in=id_list)
+        index = mapping_type.get_index(public_index)
+        if public_index:
+            qs = qs.public_indexable().privacy_level(PUBLIC)
 
-        es.flush_bulk(forced=True)
-        model.refresh_index(es=es)
+        for item in qs:
+            documents.append(mapping_type.extract_document(item.id, item))
+
+        mapping_type.bulk_index(documents, id_field='id', es=es, index=index)
+        mapping_type.refresh_index(es)
 
 
 @task
-def unindex_objects(model, ids, public_index, **kwargs):
+def unindex_objects(mapping_type, ids, public_index, **kwargs):
     if getattr(settings, 'ES_DISABLED', False):
         return
 
     es = get_es()
     for id_ in ids:
-        try:
-            model.unindex(id=id_, es=es, public_index=public_index)
-        except pyes.exceptions.ElasticSearchException, e:
-            # Patch pyes
-            if (e.status == 404 and
-                    isinstance(e.result, dict) and 'error' not in e.result):
-                # Item was not found, but command did not return an error.
-                # Do not worry.
-                return
-            else:
-                raise e
+        mapping_type.unindex(id_, es=es, public_index=public_index)
 
 
 @task

mozillians/users/tests/test_models.py

@@ -20,6 +20,7 @@ from mozillians.groups.tests import (GroupAliasFactory, GroupFactory,
                                      SkillAliasFactory, SkillFactory)
 from mozillians.users.managers import (EMPLOYEES, MOZILLIANS, PUBLIC, PUBLIC_INDEXABLE_FIELDS)
 from mozillians.users.models import ExternalAccount, UserProfile, _calculate_photo_filename, Vouch
+from mozillians.users.es import UserProfileMappingType
 from mozillians.users.tests import LanguageFactory, UserFactory
 
 
@@ -39,9 +40,9 @@ class SignaledFunctionsTests(TestCase):
                                     index_objects_mock):
         user = UserFactory.create()
         index_objects_mock.assert_called_with(
-            UserProfile, [user.userprofile.id], public_index=False)
+            UserProfileMappingType, [user.userprofile.id], public_index=False)
         unindex_objects_mock.assert_called_with(
-            UserProfile, [user.userprofile.id], public_index=True)
+            UserProfileMappingType, [user.userprofile.id], public_index=True)
 
     @patch('mozillians.users.models.index_objects.delay')
     @patch('mozillians.users.models.unindex_objects.delay')
@@ -60,8 +61,8 @@ class SignaledFunctionsTests(TestCase):
             user.delete()
 
         unindex_objects_mock.assert_has_calls([
-            call(UserProfile, [user.userprofile.id], public_index=False),
-            call(UserProfile, [user.userprofile.id], public_index=True)])
+            call(UserProfileMappingType, [user.userprofile.id], public_index=False),
+            call(UserProfileMappingType, [user.userprofile.id], public_index=True)])
 
     def test_delete_user_obj_on_profile_delete(self):
         user = UserFactory.create()
@@ -160,7 +161,7 @@ class UserProfileTests(TestCase):
         profile.skills.add(skill_1)
         profile.skills.add(skill_2)
 
-        result = UserProfile.extract_document(profile.id)
+        result = UserProfileMappingType.extract_document(profile.id)
         ok_(isinstance(result, dict))
         eq_(result['id'], profile.id)
         eq_(result['is_vouched'], profile.is_vouched)
@@ -179,25 +180,25 @@ class UserProfileTests(TestCase):
             set([u'en', u'fr', u'english', u'french', u'français']))
 
     def test_get_mapping(self):
-        ok_(UserProfile.get_mapping())
+        ok_(UserProfileMappingType.get_mapping())
 
     @override_settings(ES_INDEXES={'default': 'index'})
-    @patch('mozillians.users.models.PrivacyAwareS')
+    @patch('mozillians.users.es.PrivacyAwareS')
     def test_search_no_public_only_vouched(self, PrivacyAwareSMock):
-        result = UserProfile.search('foo')
+        result = UserProfileMappingType.search('foo')
         ok_(isinstance(result, Mock))
-        PrivacyAwareSMock.assert_any_call(UserProfile)
+        PrivacyAwareSMock.assert_any_call(UserProfileMappingType)
         PrivacyAwareSMock().indexes.assert_any_call('index')
         (PrivacyAwareSMock().indexes().boost()
          .query().order_by().filter.assert_any_call(is_vouched=True))
         ok_(call().privacy_level(PUBLIC) not in PrivacyAwareSMock.mock_calls)
 
     @override_settings(ES_INDEXES={'default': 'index'})
-    @patch('mozillians.users.models.PrivacyAwareS')
+    @patch('mozillians.users.es.PrivacyAwareS')
     def test_search_no_public_with_unvouched(self, PrivacyAwareSMock):
-        result = UserProfile.search('foo', include_non_vouched=True)
+        result = UserProfileMappingType.search('foo', include_non_vouched=True)
         ok_(isinstance(result, Mock))
-        PrivacyAwareSMock.assert_any_call(UserProfile)
+        PrivacyAwareSMock.assert_any_call(UserProfileMappingType)
         PrivacyAwareSMock().indexes.assert_any_call('index')
         ok_(call().indexes().boost()
             .query().order_by().filter(is_vouched=True)
@@ -205,11 +206,11 @@ class UserProfileTests(TestCase):
         ok_(call().privacy_level(PUBLIC) not in PrivacyAwareSMock.mock_calls)
 
     @override_settings(ES_INDEXES={'public': 'public_index'})
-    @patch('mozillians.users.models.PrivacyAwareS')
+    @patch('mozillians.users.es.PrivacyAwareS')
     def test_search_public_only_vouched(self, PrivacyAwareSMock):
-        result = UserProfile.search('foo', public=True)
+        result = UserProfileMappingType.search('foo', public=True)
         ok_(isinstance(result, Mock))
-        PrivacyAwareSMock.assert_any_call(UserProfile)
+        PrivacyAwareSMock.assert_any_call(UserProfileMappingType)
         PrivacyAwareSMock().privacy_level.assert_any_call(PUBLIC)
         (PrivacyAwareSMock().privacy_level()
          .indexes.assert_any_call('public_index'))
@@ -217,12 +218,12 @@ class UserProfileTests(TestCase):
          .query().order_by().filter.assert_any_call(is_vouched=True))
 
     @override_settings(ES_INDEXES={'public': 'public_index'})
-    @patch('mozillians.users.models.PrivacyAwareS')
+    @patch('mozillians.users.es.PrivacyAwareS')
     def test_search_public_with_unvouched(self, PrivacyAwareSMock):
-        result = UserProfile.search(
+        result = UserProfileMappingType.search(
             'foo', public=True, include_non_vouched=True)
         ok_(isinstance(result, Mock))
-        PrivacyAwareSMock.assert_any_call(UserProfile)
+        PrivacyAwareSMock.assert_any_call(UserProfileMappingType)
         PrivacyAwareSMock().privacy_level.assert_any_call(PUBLIC)
         (PrivacyAwareSMock().privacy_level()
          .indexes.assert_any_call('public_index'))
@@ -344,11 +345,11 @@ class UserProfileTests(TestCase):
 
     @override_settings(ES_INDEXES={'public': 'foo'})
     def test_get_index_public(self):
-        ok_(UserProfile.get_index(public_index=True), 'foo')
+        ok_(UserProfileMappingType.get_index(public_index=True), 'foo')
 
     @override_settings(ES_INDEXES={'default': 'bar'})
     def test_get_index(self):
-        ok_(UserProfile.get_index(public_index=False), 'bar')
+        ok_(UserProfileMappingType.get_index(public_index=False), 'bar')
 
     def test_set_privacy_level_with_save(self):
         user = UserFactory.create()

mozillians/users/tests/test_tasks.py

@@ -4,10 +4,9 @@ from django.conf import settings
 from django.contrib.auth.models import User
 from django.test.utils import override_settings
 
+from elasticsearch.exceptions import NotFoundError
 from mock import MagicMock, Mock, call, patch
-
 from nose.tools import eq_, ok_
-from pyes.exceptions import ElasticSearchException
 
 from mozillians.common.tests import TestCase
 from mozillians.groups.tests import GroupFactory
@@ -48,59 +47,57 @@ class ElasticSearchIndexTests(TestCase):
     def test_index_objects(self, get_es_mock):
         user_1 = UserFactory.create()
         user_2 = UserFactory.create()
+        mapping_type = MagicMock()
         model = MagicMock()
-        model.objects.filter.return_value = [
-            user_1.userprofile, user_2.userprofile]
-        index_objects(
-            model, [user_1.userprofile.id, user_2.userprofile.id], False)
-        model.objects.assert_has_calls([
-            call.filter(id__in=[user_1.userprofile.id, user_2.userprofile.id])])
-        model.index.assert_has_calls([
-            call(model.extract_document(), bulk=True, id_=user_1.userprofile.id,
-                 es=get_es_mock(), public_index=False),
-            call(model.extract_document(), bulk=True, id_=user_2.userprofile.id,
-                 es=get_es_mock(), public_index=False)])
-        model.refresh_index.assert_has_calls([
-            call(es=get_es_mock()),
-            call(es=get_es_mock())])
+        mapping_type.get_model.return_value = model
+        model.objects.filter.return_value = [user_1.userprofile,
+                                             user_2.userprofile]
+        mapping_type.extract_document.return_value = 'foo'
+        index_objects(mapping_type,
+                      [user_1.userprofile.id, user_2.userprofile.id],
+                      public_index=False)
+        mapping_type.bulk_index.assert_has_calls([
+            call(['foo', 'foo'], id_field='id', es=get_es_mock(),
+                 index=mapping_type.get_index(False))])
 
     @patch('mozillians.users.tasks.get_es')
     def test_index_objects_public(self, get_es_mock):
         user_1 = UserFactory.create()
         user_2 = UserFactory.create()
+        mapping_type = MagicMock()
         model = MagicMock()
-        model.objects.privacy_level().filter.return_value = [
-            user_1.userprofile, user_2.userprofile]
-        index_objects(
-            model, [user_1.userprofile.id, user_2.userprofile.id], True)
+        mapping_type.get_model.return_value = model
+        qs = model.objects.filter().public_indexable().privacy_level
+        qs.return_value = [user_1.userprofile, user_2.userprofile]
+        mapping_type.extract_document.return_value = 'foo'
+        index_objects(mapping_type,
+                      [user_1.userprofile.id, user_2.userprofile.id],
+                      public_index=True)
+
         model.objects.assert_has_calls([
-            call.filter(id__in=[user_1.userprofile.id, user_2.userprofile.id]),
-            call.privacy_level(PUBLIC)])
-        model.index.assert_has_calls([
-            call(model.extract_document(), bulk=True, id_=user_1.userprofile.id,
-                 es=get_es_mock(), public_index=True),
-            call(model.extract_document(), bulk=True, id_=user_2.userprofile.id,
-                 es=get_es_mock(), public_index=True)])
-        model.refresh_index.assert_has_calls([
-            call(es=get_es_mock()),
-            call(es=get_es_mock())])
+            call.filter(id__in=(user_1.userprofile.id, user_2.userprofile.id)),
+            call.filter().public_indexable(),
+            call.filter().public_indexable().privacy_level(PUBLIC),
+        ])
+        mapping_type.bulk_index.assert_has_calls([
+            call(['foo', 'foo'], id_field='id', es=get_es_mock(),
+                 index=mapping_type.get_index(True))])
 
     @patch('mozillians.users.tasks.get_es')
     def test_unindex_objects(self, get_es_mock):
-        model = MagicMock()
-        unindex_objects(model, [1, 2, 3], 'foo')
-        ok_(model.unindex.called)
-        model.assert_has_calls([
-            call.unindex(es=get_es_mock(), public_index='foo', id=1),
-            call.unindex(es=get_es_mock(), public_index='foo', id=2),
-            call.unindex(es=get_es_mock(), public_index='foo', id=3)])
+        mapping_type = MagicMock()
+        unindex_objects(mapping_type, [1, 2, 3], 'foo')
+        ok_(mapping_type.unindex.called)
+        mapping_type.assert_has_calls([
+            call.unindex(1, es=get_es_mock(), public_index='foo'),
+            call.unindex(2, es=get_es_mock(), public_index='foo'),
+            call.unindex(3, es=get_es_mock(), public_index='foo')])
 
     def test_unindex_raises_not_found_exception(self):
-        exception = ElasticSearchException(
-            error=404, status=404, result={'not found': 'not found'})
-        model = Mock()
-        model.unindex = Mock(side_effect=exception)
-        unindex_objects(model, [1, 2, 3], 'foo')
+        exception = NotFoundError(404, {'not found': 'not found '}, {'foo': 'foo'})
+        mapping_type = Mock()
+        mapping_type.unindex(side_effect=exception)
+        unindex_objects(mapping_type, [1, 2, 3], 'foo')
 
 
 class BasketTests(TestCase):

vendor-local/src/pyes

@@ -1 +0,0 @@
-Subproject commit 27d00eac9030cc9c4dfce9231ad1094f1470a3ca

vendor-local/vendor.pth

@@ -3,7 +3,6 @@ src/mimeparse
 src/basket-client-submodule
 src/django-autocomplete-light
 src/elasticutils
-src/pyes
 src/django-tastypie
 src/pystatsd
 src/happyforms