# -*- 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 specific hooks and operations""" import logging from elasticsearch import Elasticsearch from elasticsearch.exceptions import ConnectionError from urllib3.exceptions import ProtocolError from cubicweb.server import hook from cubicweb.predicates import score_entity from cubes.elasticsearch.es import indexable_types log = logging.getLogger(__name__) def entity_indexable(entity): return entity.cw_etype in indexable_types(entity._cw.vreg.schema) class ContentUpdateIndexES(hook.Hook): """detect content change and updates ES indexing""" __regid__ = 'elasticsearch.contentupdatetoes' __select__ = hook.Hook.__select__ & score_entity(entity_indexable) events = ('after_update_entity', 'after_add_entity') category = 'es' def __call__(self): if self.entity.cw_etype == 'File': return # FIXME hack! locations = self._cw.vreg.config['elasticsearch-locations'] index_name = self._cw.vreg.config['index-name'] if locations and index_name: es = Elasticsearch(locations and locations.split(',')) # TODO : have a cache so that we don't check index on every transition if not es.indices.exists(index=index_name): es.indices.create(index=index_name, body=INDEX_SETTINGS, ignore=400) # serializer = self.entity.cw_adapt_to('IFTISerializable') serializer = self.entity.cw_adapt_to('ISerializable') json = serializer.serialize() try: # TODO option pour coté async ? es.index(index=index_name, id=self.entity.eid, doc_type=self.entity.cw_etype, body=json) except (ConnectionError, ProtocolError): log.debug('Failed to index in hook, could not connect to ES')