Merge pull request #42 from brandicted/develop

release 0.2.3

docs/source/changelog.rst

@@ -1,10 +1,17 @@
 Changelog
 =========
 
+* :release:`0.2.3 <2015-06-03>`
+* :bug:`-` Fixed password minimum length support by adding before and after validation processors
+* :bug:`-` Fixed bug with Elasticsearch indexing of nested relationships
+* :bug:`-` Fixed race condition in Elasticsearch indexing
+
 * :release:`0.2.2 <2015-05-27>`
-* :bug:`-` fixes login issue
-* :bug:`-` fixes posting to singular resources e.g. /api/users/<username>/profile
-* :bug:`-` fixes multiple foreign keys to same model
+* :bug:`-` Fixed login issue
+* :bug:`-` Fixed posting to singular resources e.g. /api/users/<username>/profile
+* :bug:`-` Fixed multiple foreign keys to same model
+* :bug:`-` Fixed ES mapping error when values of field were all null
+* :bug:`-` Fixed a bug whereby Relationship could not be created without a backref
 
 * :release:`0.2.1 <2015-05-20>`
 * :bug:`-` Fixed slow queries to backrefs

nefertari_sqla/documents.py

@@ -7,6 +7,7 @@
 from sqlalchemy.orm.collections import InstrumentedList
 from sqlalchemy.exc import InvalidRequestError, IntegrityError
 from sqlalchemy.orm.exc import MultipleResultsFound, NoResultFound
+from sqlalchemy.orm.query import Query
 from sqlalchemy.orm.properties import RelationshipProperty
 from pyramid_sqlalchemy import Session, BaseObject
 
@@ -15,7 +16,7 @@
 from nefertari.utils import (
     process_fields, process_limit, _split, dictset,
     DataProxy)
-from .signals import ESMetaclass
+from .signals import ESMetaclass, on_bulk_delete
 from .fields import ListField, DictField, DateTimeField, IntegerField
 from . import types
 
@@ -67,23 +68,23 @@ def process_bools(_dict):
     types.LimitedText: {'type': 'string'},
     types.LimitedUnicode: {'type': 'string'},
     types.LimitedUnicodeText: {'type': 'string'},
-    types.ProcessableChoice: {'type': 'string'},
+    types.Choice: {'type': 'string'},
 
-    types.ProcessableBoolean: {'type': 'boolean'},
-    types.ProcessableLargeBinary: {'type': 'object'},
-    types.ProcessableDict: {'type': 'object'},
+    types.Boolean: {'type': 'boolean'},
+    types.LargeBinary: {'type': 'object'},
+    types.Dict: {'type': 'object'},
 
     types.LimitedNumeric: {'type': 'double'},
     types.LimitedFloat: {'type': 'double'},
 
     types.LimitedInteger: {'type': 'long'},
     types.LimitedBigInteger: {'type': 'long'},
     types.LimitedSmallInteger: {'type': 'long'},
-    types.ProcessableInterval: {'type': 'long'},
+    types.Interval: {'type': 'long'},
 
-    types.ProcessableDateTime: {'type': 'date', 'format': 'dateOptionalTime'},
-    types.ProcessableDate: {'type': 'date', 'format': 'dateOptionalTime'},
-    types.ProcessableTime: {'type': 'date', 'format': 'HH:mm:ss'},
+    types.DateTime: {'type': 'date', 'format': 'dateOptionalTime'},
+    types.Date: {'type': 'date', 'format': 'dateOptionalTime'},
+    types.Time: {'type': 'date', 'format': 'HH:mm:ss'},
 }
 
 
@@ -119,18 +120,27 @@ def get_es_mapping(cls):
         }
         mapper = class_mapper(cls)
         columns = {c.name: c for c in mapper.columns}
+        relationships = {r.key: r for r in mapper.relationships}
         # Replace field 'id' with primary key field
         columns['id'] = columns.get(cls.pk_field())
 
         for name, column in columns.items():
             column_type = column.type
-            if isinstance(column_type, types.ProcessableChoiceArray):
+            if isinstance(column_type, types.ChoiceArray):
                 column_type = column_type.impl.item_type
             column_type = type(column_type)
             if column_type not in TYPES_MAP:
                 continue
             properties[name] = TYPES_MAP[column_type]
 
+        for name, column in relationships.items():
+            if name in cls._nested_relationships:
+                column_type = {'type': 'object'}
+            else:
+                rel_pk_field = column.mapper.class_.pk_field_type()
+                column_type = TYPES_MAP[rel_pk_field]
+            properties[name] = column_type
+
         properties['_type'] = {'type': 'string'}
         return mapping
 
@@ -241,10 +251,9 @@ def filter_objects(cls, objects, first=False, **params):
         if first:
             params['_limit'] = 1
             params['__raise_on_empty'] = True
-        params['query_set'] = query_set.from_self()
-        query_set = cls.get_collection(**params)
+            params['query_set'] = query_set.from_self()
+            query_set = cls.get_collection(**params)
 
-        if first:
             first_obj = query_set.first()
             if not first_obj:
                 msg = "'{}({}={})' resource not found".format(
@@ -439,8 +448,7 @@ def get_or_create(cls, **params):
         except NoResultFound:
             defaults.update(params)
             new_obj = cls(**defaults)
-            query_set.session.add(new_obj)
-            query_set.session.flush()
+            new_obj.save()
             return new_obj, True
         except MultipleResultsFound:
             raise JHTTPBadRequest('Bad or Insufficient Params')
@@ -465,21 +473,58 @@ def _update(self, params, **kw):
         return self
 
     @classmethod
-    def _delete(cls, **params):
-        obj = cls.get(**params)
-        object_session(obj).delete(obj)
+    def _delete_many(cls, items, synchronize_session=False,
+                     refresh_index=None):
+        """ Delete :items: queryset or objects list.
 
-    @classmethod
-    def _delete_many(cls, items):
+        When queryset passed, Query.delete() is used to delete it. Note that
+        queryset may not have limit(), offset(), order_by(), group_by(), or
+        distinct() called on it.
+
+        If some of the methods listed above were called, or :items: is not
+        a Query instance, one-by-one items update is performed.
+
+        `on_bulk_delete` function is called to delete objects from index
+        and to reindex relationships. This is done explicitly because it is
+        impossible to get access to deleted objects in signal handler for
+        'after_bulk_delete' ORM event.
+        """
+        if isinstance(items, Query):
+            try:
+                delete_items = items.all()
+                items.delete(
+                    synchronize_session=synchronize_session)
+                on_bulk_delete(cls, delete_items, refresh_index=refresh_index)
+                return
+            except Exception as ex:
+                log.error(str(ex))
         session = Session()
         for item in items:
+            item._refresh_index = refresh_index
             session.delete(item)
         session.flush()
 
     @classmethod
-    def _update_many(cls, items, **params):
+    def _update_many(cls, items, synchronize_session='fetch',
+                     refresh_index=None, **params):
+        """ Update :items: queryset or objects list.
+
+        When queryset passed, Query.update() is used to update it. Note that
+        queryset may not have limit(), offset(), order_by(), group_by(), or
+        distinct() called on it.
+
+        If some of the methods listed above were called, or :items: is not
+        a Query instance, one-by-one items update is performed.
+        """
+        if isinstance(items, Query):
+            try:
+                items._refresh_index = refresh_index
+                return items.update(
+                    params, synchronize_session=synchronize_session)
+            except Exception as ex:
+                log.error(str(ex))
         for item in items:
-            item.update(params)
+            item.update(params, refresh_index=refresh_index)
 
     def __repr__(self):
         parts = []
@@ -532,7 +577,9 @@ def to_dict(self, **kwargs):
         return _dict
 
     def update_iterables(self, params, attr, unique=False,
-                         value_type=None, save=True):
+                         value_type=None, save=True,
+                         refresh_index=None):
+        self._refresh_index = refresh_index
         mapper = class_mapper(self.__class__)
         columns = {c.name: c for c in mapper.columns}
         is_dict = isinstance(columns.get(attr), DictField)
@@ -568,9 +615,7 @@ def update_dict(update_params):
 
             setattr(self, attr, final_value)
             if save:
-                session = object_session(self)
-                session.add(self)
-                session.flush()
+                self.save(refresh_index=refresh_index)
 
         def update_list(update_params):
             final_value = getattr(self, attr, []) or []
@@ -594,9 +639,7 @@ def update_list(update_params):
 
             setattr(self, attr, final_value)
             if save:
-                session = object_session(self)
-                session.add(self)
-                session.flush()
+                self.save(refresh_index=refresh_index)
 
         if is_dict:
             update_dict(params)
@@ -615,8 +658,11 @@ def get_reference_documents(self):
             # If 'Many' side should be indexed, its value is already a list.
             if value is None or isinstance(value, list):
                 continue
-            session = object_session(value)
-            session.refresh(value)
+            try:
+                session = object_session(value)
+                session.refresh(value)
+            except InvalidRequestError:
+                pass
             yield (value.__class__, [value.to_dict()])
 
     def _is_modified(self):
@@ -651,15 +697,17 @@ def _bump_version(self):
             self.updated_at = datetime.utcnow()
             self._version = (self._version or 0) + 1
 
-    def save(self, *arg, **kw):
+    def save(self, refresh_index=None):
         session = object_session(self)
         self._bump_version()
+        self._refresh_index = refresh_index
         session = session or Session()
         try:
-            self.clean()
+            self.apply_before_validation()
             session.add(self)
             session.flush()
             session.expire(self)
+            self.apply_after_validation()
             return self
         except (IntegrityError,) as e:
             if 'duplicate' not in e.message:
@@ -670,14 +718,16 @@ def save(self, *arg, **kw):
                     self.__class__.__name__),
                 extra={'data': e})
 
-    def update(self, params):
+    def update(self, params, refresh_index=None):
+        self._refresh_index = refresh_index
         try:
             self._update(params)
             self._bump_version()
-            self.clean()
+            self.apply_before_validation()
             session = object_session(self)
             session.add(self)
             session.flush()
+            self.apply_after_validation()
             return self
         except (IntegrityError,) as e:
             if 'duplicate' not in e.message:
@@ -688,9 +738,36 @@ def update(self, params):
                     self.__class__.__name__),
                 extra={'data': e})
 
-    def clean(self, force_all=False):
-        """ Apply field processors to all changed fields And perform custom
-        field values cleaning before running DB validation.
+    def delete(self, refresh_index=None):
+        self._refresh_index = refresh_index
+        object_session(self).delete(self)
+
+    def apply_processors(self, column_names=None, before=False, after=False):
+        """ Apply processors to columns with :column_names: names.
+
+        Arguments:
+          :column_names: List of string names of changed columns.
+          :before: Boolean indicating whether to apply before_validation
+            processors.
+          :after: Boolean indicating whether to apply after_validation
+            processors.
+        """
+        columns = {c.key: c for c in class_mapper(self.__class__).columns}
+        if column_names is None:
+            column_names = columns.keys()
+
+        for name in column_names:
+            column = columns.get(name)
+            if column is not None and hasattr(column, 'apply_processors'):
+                new_value = getattr(self, name)
+                processed_value = column.apply_processors(
+                    instance=self, new_value=new_value,
+                    before=before, after=after)
+                setattr(self, name, processed_value)
+
+    def apply_before_validation(self):
+        """ Determine changed columns and run `self.apply_processors` to
+        apply needed processors.
 
         Note that at this stage, field values are in the exact same state
         you posted/set them. E.g. if you set time_field='11/22/2000',
@@ -699,18 +776,23 @@ def clean(self, force_all=False):
         columns = {c.key: c for c in class_mapper(self.__class__).columns}
         state = attributes.instance_state(self)
 
-        if state.persistent and not force_all:
+        if state.persistent:
             changed_columns = state.committed_state.keys()
         else:  # New object
             changed_columns = columns.keys()
 
-        for name in changed_columns:
-            column = columns.get(name)
-            if column is not None and hasattr(column, 'apply_processors'):
-                new_value = getattr(self, name)
-                processed_value = column.apply_processors(
-                    instance=self, new_value=new_value)
-                setattr(self, name, processed_value)
+        self._columns_to_process = changed_columns
+        self.apply_processors(changed_columns, before=True)
+
+    def apply_after_validation(self):
+        """ Run `self.apply_processors` with columns names determined by
+        `self.apply_before_validation`.
+
+        Note that at this stage, field values are in the exact same state
+        you posted/set them. E.g. if you set time_field='11/22/2000',
+        self.time_field will be equal to '11/22/2000' here.
+        """
+        self.apply_processors(self._columns_to_process, after=True)
 
 
 class ESBaseDocument(BaseDocument):