# -*- coding: utf-8 -*-
# copyright 2016 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
# contact http://www.logilab.fr -- mailto:contact@logilab.fr
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 2.1 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""cubicweb-elasticsearch entity's classes"""
import collections
from functools import partial
from urllib3.exceptions import ProtocolError
from elasticsearch.exceptions import ConnectionError, NotFoundError
from logilab.common.decorators import cachedproperty
from cubicweb import view, neg_role
from cubicweb.predicates import is_instance
from cubicweb.appobject import AppObject
from cubicweb_elasticsearch import es
def deep_update(d1, d2):
for key, value in d2.iteritems():
if isinstance(value, collections.Mapping):
d1[key] = deep_update(d1.get(key, {}), value)
else:
d1[key] = d2[key]
return d1
class EsRegistry(AppObject):
__registry__ = 'es'
class Indexer(EsRegistry):
__regid__ = 'indexer'
settings = {
'settings': {
'analysis': {
'analyzer': {
'default': {'filter': ['standard',
'my_ascii_folding',
'lowercase',
'french_snowball'],
'tokenizer': 'standard'}
},
'filter': {'my_ascii_folding': {'preserve_original': True,
'type': 'asciifolding'},
'french_snowball': {'type': 'snowball',
'language': 'French'}}
},
}
}
@property
def index_name(self):
return self._cw.vreg.config['index-name']
def get_connection(self):
self.create_index()
return es.get_connection(self._cw.vreg.config)
def create_index(self, index_name=None, custom_settings=None):
index_name = index_name or self.index_name
if custom_settings is None:
settings = self.settings
else:
settings = {}
deep_update(settings, self.settings)
deep_update(settings, custom_settings)
es_cnx = es.get_connection(self._cw.vreg.config)
if es_cnx is not None:
es.create_index(es_cnx, index_name, settings)
def es_index(self, entity, params=None):
es_cnx = self.get_connection()
if es_cnx is None or not self.index_name:
self.error('no connection to ES (not configured) skip ES indexing')
return
serializable = entity.cw_adapt_to('IFullTextIndexSerializable')
json = serializable.serialize()
if not json:
return
es_cnx.index(index=self.index_name, id=serializable.es_id,
doc_type=serializable.es_doc_type, body=json,
params=params)
def es_delete(self, entity):
es_cnx = self.get_connection()
if es_cnx is None or not self.index_name:
self.error('no connection to ES (not configured) skip ES deletion')
return
serializable = entity.cw_adapt_to('IFullTextIndexSerializable')
es_cnx.delete(index=self.index_name, id=serializable.es_id,
doc_type=serializable.es_doc_type)
class IFullTextIndexSerializable(view.EntityAdapter):
"""Adapter to serialize an entity to a bare python structure that may be
directly serialized to e.g. JSON.
"""
__regid__ = 'IFullTextIndexSerializable'
__select__ = is_instance('Any')
custom_indexable_attributes = ()
skip_indexable_attributes = ()
@property
def es_id(self):
return self.entity.eid
@property
def es_doc_type(self):
return self.entity.cw_etype
@cachedproperty
def fulltext_indexable_attributes(self):
eschema = self._cw.vreg.schema[self.entity.cw_etype]
attrs = ['creation_date', 'modification_date', 'cwuri']
attrs.extend([r.type for r in eschema.indexable_attributes()
if r.type not in self.skip_indexable_attributes])
for rschema, tschema in eschema.attribute_definitions():
if rschema.type == 'eid':
continue
# XXX
if tschema.type in ('Int', 'Float'):
attrs.append(rschema.type)
attrs.extend(self.custom_indexable_attributes)
return attrs
def process_attributes(self):
data = {}
for attr in self.fulltext_indexable_attributes:
data[attr] = getattr(self.entity, attr)
return data
def serialize(self, complete=True):
entity = self.entity
if complete:
entity.complete()
data = {
'cw_etype': entity.cw_etype,
'eid': entity.eid,
'cwuri': entity.cwuri,
}
# TODO take a look at what's in entity.cw_relation_cache
data.update(self.process_attributes())
return data
class File(IFullTextIndexSerializable):
__select__ = IFullTextIndexSerializable.__select__ & is_instance('File')
def serialize(self, complete=True):
"""this could be a generic implementation of fulltext_containers indexation, but for
now we can not return more than one parent json which is fine
for Files
"""
for rschema, role in self._cw.vreg.schema['File'].fulltext_containers():
for parent in self.entity.related(
rschema.type, role=neg_role(role)).entities():
return parent.cw_adapt_to(
'IFullTextIndexSerializable').serialize(complete)
return {}
class ESTransactionQueue(EsRegistry):
__regid__ = 'es.opqueue'
@cachedproperty
def default_indexer(self):
return self._cw.vreg['es'].select('indexer', self._cw)
def purge_useless_operations(self, es_operations):
"""remove operations from `es_operations` that will have no effect.
For instance:
- if an entity is indexed several times, just index it once.
- if an entity is indexed then deleted, simply remove the
*index* operation.
Since there are only two kind of operations ('delete' and 'index'),
the general rule is that the last operation wins and therefore the
first one is useless. The only corner case is when a delete operation
is triggered before an index one. In that specific case, emit a warning
and keep delete operation only.
"""
done = collections.OrderedDict()
for es_operation in reversed(es_operations):
entity = es_operation['entity']
op_type = es_operation['op_type']
if entity.eid not in done:
done[entity.eid] = es_operation
else:
prev_op_type = done[entity.eid]['op_type']
if op_type == 'delete' and prev_op_type == 'index':
self.warning('a delete operation on %s#%s inserted before'
'an index one', entity.cw_etype, entity.eid)
done[entity.eid] = es_operation
return done.values()
def process_operation(self, es_operation):
indexer = es_operation.get('indexer', self.default_indexer)
entity = es_operation['entity']
if self._cw.deleted_in_transaction(entity.eid):
es_method = indexer.es_delete
elif es_operation['op_type'] == 'index':
es_method = partial(indexer.es_index, params={'refresh': True})
elif es_operation['op_type'] == 'delete':
es_method = indexer.es_delete
else:
self.info('skipping unknown operation type %s on %s',
es_operation['op_type'], entity.eid)
return
try:
es_method(entity)
except (ConnectionError, ProtocolError, NotFoundError) as exc:
self.warning('[ES] Failed to %s %s#%s (%s)',
es_operation['op_type'],
entity.cw_etype, entity.eid, exc)
def process_operations(self, es_operations):
es_operations = self.purge_useless_operations(es_operations)
for es_operation in es_operations:
self.process_operation(es_operation)
return es_operations