cubicweb: comparison server/sources/datafeed.py

equal deleted inserted replaced

-:928732ec00dd
+:fa44db7da2dc
 def init(self, activated, source_entity):
 super(DataFeedSource, self).init(activated, source_entity)
 self.parser_id = source_entity.parser
 self.load_mapping(source_entity._cw)
-def _get_parser(self, session, **kwargs):
+def _get_parser(self, cnx, **kwargs):
 return self.repo.vreg['parsers'].select(
-self.parser_id, session, source=self, **kwargs)
+self.parser_id, cnx, source=self, **kwargs)
-def load_mapping(self, session):
+def load_mapping(self, cnx):
 self.mapping = {}
 self.mapping_idx = {}
 try:
-parser = self._get_parser(session)
+parser = self._get_parser(cnx)
 except (RegistryNotFound, ObjectNotFound):
 return # no parser yet, don't go further
-self._load_mapping(session, parser=parser)
+self._load_mapping(cnx, parser=parser)
 def add_schema_config(self, schemacfg, checkonly=False, parser=None):
 """added CWSourceSchemaConfig, modify mapping accordingly"""
 if parser is None:
 parser = self._get_parser(schemacfg._cw)
 return datetime.utcnow() < (self.latest_retrieval + self.synchro_interval)
 def update_latest_retrieval(self, cnx):
 self.latest_retrieval = datetime.utcnow()
 cnx.execute('SET X latest_retrieval %(date)s WHERE X eid %(x)s',
 {'x': self.eid, 'date': self.latest_retrieval})
 cnx.commit()
 def acquire_synchronization_lock(self, cnx):
 # XXX race condition until WHERE of SET queries is executed using
 # 'SELECT FOR UPDATE'
 cnx.commit()
 return True
 def release_synchronization_lock(self, cnx):
 cnx.execute('SET X in_synchronization NULL WHERE X eid %(x)s',
 {'x': self.eid})
 cnx.commit()
 def pull_data(self, cnx, force=False, raise_on_error=False):
 """Launch synchronization of the source if needed.
 self.exception('error while processing %s: %s',
 url, exc)
 error = True
 return error
-def before_entity_insertion(self, session, lid, etype, eid, sourceparams):
+def before_entity_insertion(self, cnx, lid, etype, eid, sourceparams):
 """called by the repository when an eid has been attributed for an
 entity stored here but the entity has not been inserted in the system
 table yet.
 This method must return the an Entity instance representation of this
 entity.
 """
 entity = super(DataFeedSource, self).before_entity_insertion(
-session, lid, etype, eid, sourceparams)
+cnx, lid, etype, eid, sourceparams)
 entity.cw_edited['cwuri'] = lid.decode('utf-8')
 entity.cw_edited.set_defaults()
 sourceparams['parser'].before_entity_copy(entity, sourceparams)
 return entity
-def after_entity_insertion(self, session, lid, entity, sourceparams):
+def after_entity_insertion(self, cnx, lid, entity, sourceparams):
 """called by the repository after an entity stored here has been
 inserted in the system table.
 """
-relations = preprocess_inlined_relations(session, entity)
+relations = preprocess_inlined_relations(cnx, entity)
-if session.is_hook_category_activated('integrity'):
+if cnx.is_hook_category_activated('integrity'):
 entity.cw_edited.check(creation=True)
-self.repo.system_source.add_entity(session, entity)
+self.repo.system_source.add_entity(cnx, entity)
 entity.cw_edited.saved = entity._cw_is_saved = True
 sourceparams['parser'].after_entity_copy(entity, sourceparams)
 # call hooks for inlined relations
 call_hooks = self.repo.hm.call_hooks
 if self.should_call_hooks:
 for attr, value in relations:
-call_hooks('before_add_relation', session,
+call_hooks('before_add_relation', cnx,
 eidfrom=entity.eid, rtype=attr, eidto=value)
-call_hooks('after_add_relation', session,
+call_hooks('after_add_relation', cnx,
 eidfrom=entity.eid, rtype=attr, eidto=value)
-def source_cwuris(self, session):
+def source_cwuris(self, cnx):
 sql = ('SELECT extid, eid, type FROM entities, cw_source_relation '
 'WHERE entities.eid=cw_source_relation.eid_from '
 'AND cw_source_relation.eid_to=%s' % self.eid)
 return dict((b64decode(uri), (eid, type))
-for uri, eid, type in session.system_sql(sql).fetchall())
+for uri, eid, type in cnx.system_sql(sql).fetchall())
-def init_import_log(self, session, **kwargs):
+def init_import_log(self, cnx, **kwargs):
-dataimport = session.create_entity('CWDataImport', cw_import_of=self,
+dataimport = cnx.create_entity('CWDataImport', cw_import_of=self,
 start_timestamp=datetime.utcnow(),
 **kwargs)
 dataimport.init()
 return dataimport
 class DataFeedParser(AppObject):
 __registry__ = 'parsers'
-def __init__(self, session, source, sourceuris=None, import_log=None, **kwargs):
+def __init__(self, cnx, source, sourceuris=None, import_log=None, **kwargs):
-super(DataFeedParser, self).__init__(session, **kwargs)
+super(DataFeedParser, self).__init__(cnx, **kwargs)
 self.source = source
 self.sourceuris = sourceuris
 self.import_log = import_log
 self.stats = {'created': set(), 'updated': set(), 'checked': set()}
 def extid2entity(self, uri, etype, **sourceparams):
 """return an entity for the given uri. May return None if it should be
 skipped
 """
-session = self._cw
+cnx = self._cw
 # if cwsource is specified and repository has a source with the same
 # name, call extid2eid on that source so entity will be properly seen as
 # coming from this source
 source_uri = sourceparams.pop('cwsource', None)
 if source_uri is not None and source_uri != 'system':
-source = session.repo.sources_by_uri.get(source_uri, self.source)
+source = cnx.repo.sources_by_uri.get(source_uri, self.source)
 else:
 source = self.source
 sourceparams['parser'] = self
 if isinstance(uri, unicode):
 uri = uri.encode('utf-8')
 try:
-eid = session.repo.extid2eid(source, str(uri), etype, session,
+eid = cnx.repo.extid2eid(source, str(uri), etype, cnx,
 sourceparams=sourceparams)
 except ValidationError as ex:
 # XXX use critical so they are seen during tests. Should consider
 # raise_on_error instead?
 self.source.critical('error while creating %s: %s', etype, ex)
 # entity has been moved away from its original source
 #
 # Don't give etype to entity_from_eid so we get UnknownEid if the
 # entity has been removed
 try:
-entity = session.entity_from_eid(-eid)
+entity = cnx.entity_from_eid(-eid)
 except UnknownEid:
 return None
 self.notify_updated(entity) # avoid later update from the source's data
 return entity
 if self.sourceuris is not None:
 self.sourceuris.pop(str(uri), None)
-return session.entity_from_eid(eid, etype)
+return cnx.entity_from_eid(eid, etype)
 def process(self, url, raise_on_error=False):
 """main callback: process the url"""
 raise NotImplementedError
 is actually deleted. Always return True by default, put more sensible
 stuff in sub-classes.
 """
 return True
-def handle_deletion(self, config, session, myuris):
+def handle_deletion(self, config, cnx, myuris):
 if config['delete-entities'] and myuris:
 byetype = {}
 for extid, (eid, etype) in myuris.iteritems():
 if self.is_deleted(extid, etype, eid):
 byetype.setdefault(etype, []).append(str(eid))
 for etype, eids in byetype.iteritems():
 self.warning('delete %s %s entities', len(eids), etype)
-session.set_cnxset()
+cnx.execute('DELETE %s X WHERE X eid IN (%s)'
-session.execute('DELETE %s X WHERE X eid IN (%s)'
+% (etype, ','.join(eids)))
-% (etype, ','.join(eids)))
+cnx.commit()
-session.commit()
 def update_if_necessary(self, entity, attrs):
 entity.complete(tuple(attrs))
 # check modification date and compare attribute values to only update
 # what's actually needed
 if raise_on_error:
 raise
 self.import_log.record_error(str(ex))
 return True
 error = False
-# Check whether self._cw is a session or a connection
+commit = self._cw.commit
-if getattr(self._cw, 'commit', None) is not None:
+rollback = self._cw.rollback
-commit = self._cw.commit
-rollback = self._cw.rollback
-else:
-commit = self._cw.cnx.commit
-rollback = self._cw.cnx.rollback
 for args in parsed:
 try:
 self.process_item(*args)
 # commit+set_cnxset instead of commit(free_cnxset=False) to let
 # other a chance to get our connections set

changeset 9897	fa44db7da2dc
parent 9825	946b483bc8a1
parent 9879	21278eb03bbf
child 9990	c84ad981fc4a