cubicweb: comparison server/repository.py

equal deleted inserted replaced

-:8ea54e7be3e2
+:93c613913912
 def commit_event(self):
 """the observed connections pool has been rollbacked,
 remove inserted eid from repository type/source cache
 """
 self.repo.clear_caches(self.session.query_data('pendingeids', ()))
 def rollback_event(self):
 """the observed connections pool has been rollbacked,
 remove inserted eid from repository type/source cache
 """
 self.repo.clear_caches(self.session.query_data('neweids', ()))
 if entity.eid in session.query_data('pendingeids', ()):
 return # entity added and deleted in the same transaction
 session.repo.system_source.fti_unindex_entity(session, entity.eid)
 for container in entity.fti_containers():
 session.repo.index_entity(session, container)
 def commit_event(self):
 pass
 def del_existing_rel_if_needed(session, eidfrom, rtype, eidto):
 """delete existing relation when adding a new one if card is 1 or ?
 if card[1] in '1?':
 session.unsafe_execute(
 'DELETE X %s Y WHERE NOT X eid %%(x)s, Y eid %%(y)s' % rtype,
 {'x': eidfrom, 'y': eidto}, 'y')
 class Repository(object):
 """a repository provides access to a set of persistent storages for
 entities and relations
 XXX protect pyro access
 """
 def __init__(self, config, vreg=None, debug=False):
 self.config = config
 if vreg is None:
 vreg = CubicWebRegistry(config, debug)
 self.vreg = vreg
 # FIXME: store additional sources info in the system database ?
 # FIXME: sources should be ordered (add_entity priority)
 for uri, source_config in config.sources().items():
 if uri == 'admin':
 # not an actual source
 continue
 source = self.get_source(uri, source_config)
 self.sources_by_uri[uri] = source
 self.sources.append(source)
 self.system_source = self.sources_by_uri['system']
 # ensure system source is the first one
 # tsearch according to postgres version
 for source in self.sources:
 source.init_creating()
 # close initialization pool and reopen fresh ones for proper
 # initialization now that we know cubes
 self._get_pool().close(True)
 for i in xrange(config['connections-pool-size']):
 self._available_pools.put_nowait(ConnectionsPool(self.sources))
 # internals ###############################################################
 def get_source(self, uri, source_config):
 source_config['uri'] = uri
 return get_source(source_config, self.schema, self)
 def set_schema(self, schema, resetvreg=True):
 schema.rebuild_infered_relations()
 self.info('set schema %s %#x', schema.name, id(schema))
 self.debug(', '.join(sorted(str(e) for e in schema.entities())))
 self.querier.set_schema(schema)
 except BadSchemaDefinition:
 raise
 except Exception, ex:
 import traceback
 traceback.print_exc()
 raise Exception('Is the database initialised ? (cause: %s)' %
 (ex.args and ex.args[0].strip() or 'unknown')), \
 None, sys.exc_info()[-1]
 self.info('set the actual schema')
 # XXX have to do this since CWProperty isn't in the bootstrap schema
 #     it'll be redone in set_schema
 session.commit()
 finally:
 session.close()
 self.config.init_cubes(self.get_cubes())
 self.set_schema(appschema)
 def set_bootstrap_schema(self, schema):
 """disable hooks when setting a bootstrap schema, but restore
 the configuration for the next time
 """
 config = self.config
 config.core_hooks = True
 config.usergroup_hooks = True
 config.schema_hooks = True
 config.notification_hooks = True
 config.application_hooks = True
 def start_looping_tasks(self):
 assert isinstance(self._looping_tasks, list), 'already started'
 for i, (interval, func) in enumerate(self._looping_tasks):
 self._looping_tasks[i] = task = LoopTask(interval, func)
 self.info('starting task %s with interval %.2fs', task.name,
 # ensure no tasks will be further added
 self._looping_tasks = tuple(self._looping_tasks)
 def looping_task(self, interval, func):
 """register a function to be called every `interval` seconds.
 looping tasks can only be registered during repository initialization,
 once done this method will fail.
 """
 try:
 self._looping_tasks.append( (interval, func) )
 def threaded_task(self, func):
 """start function in a separated thread"""
 t = RepoThread(func, self._running_threads)
 t.start()
 #@locked
 def _get_pool(self):
 try:
 return self._available_pools.get(True, timeout=5)
 except Queue.Empty:
 raise Exception('no pool available after 5 secs, probably either a '
 'bug in code (to many uncommited/rollbacked '
 'connections) or to much load on the server (in '
 'which case you can try to set a bigger '
 'connections pools size)')
 def _free_pool(self, pool):
 pool.rollback()
 self._available_pools.put_nowait(pool)
 def pinfo(self):
 nocache  = self.system_source.no_cache
 self.info('sql cache usage: %s/%s (%s%%)', hits+ misses, nocache,
 ((hits + misses) * 100) / (hits + misses + nocache))
 except ZeroDivisionError:
 pass
 def authenticate_user(self, session, login, password):
 """validate login / password, raise AuthenticationError on failure
 return associated CWUser instance on success
 """
 for source in self.sources:
 # prefetch / cache euser's groups and properties. This is especially
 # useful for internal sessions to avoid security insertions
 euser.groups
 euser.properties
 return euser
 # public (dbapi) interface ################################################
 def get_schema(self):
 """return the application schema. This is a public method, not
 requiring a session id
 """
 try:
 'is installed. Run "cubicweb-ctl upgrade".')
 raise ExecutionError(msg % (cube, version, fsversion))
 finally:
 session.close()
 return vcconf
 @cached
 def source_defs(self):
 sources = self.config.sources().copy()
 # remove manager information
 sources.pop('admin', None)
 'U primary_email X, U use_email X WHERE U login %(login)s', d)
 session.commit()
 finally:
 session.close()
 return True
 def connect(self, login, password, cnxprops=None):
 """open a connection for a given user
 base_url may be needed to send mails
 cnxtype indicate if this is a pyro connection or a in-memory connection
 raise `AuthenticationError` if the authentication failed
 raise `ConnectionError` if we can't open a connection
 """
 # use an internal connection
 session = self.internal_session()
 # FIXME: check error to catch internal errors
 self.exception('unexpected error')
 raise
 finally:
 session.reset_pool()
 def describe(self, sessionid, eid):
 """return a tuple (type, source, extid) for the entity with id <eid>"""
 session = self._get_session(sessionid, setpool=True)
 try:
 return self.type_and_source_from_eid(eid, session)
 def commit(self, sessionid):
 """commit transaction for the session with the given id"""
 self.debug('begin commit for session %s', sessionid)
 try:
 self._get_session(sessionid, setpool=True).commit()
 except (ValidationError, Unauthorized):
 raise
 except:
 self.exception('unexpected error')
 raise
 def rollback(self, sessionid):
 """commit transaction for the session with the given id"""
 self.debug('begin rollback for session %s', sessionid)
 try:
 self._get_session(sessionid, setpool=True).rollback()
 # during `session_close` hooks
 session.commit()
 session.close()
 del self._sessions[sessionid]
 self.info('closed session %s for user %s', sessionid, session.user.login)
 def user_info(self, sessionid, props=None):
 """this method should be used by client to:
 * check session id validity
 * update user information on each user's request (i.e. groups and
 custom properties)
 # update session properties
 for prop, value in props.items():
 session.change_property(prop, value)
 user = session.user
 return user.eid, user.login, user.groups, user.properties
 # public (inter-repository) interface #####################################
 def entities_modified_since(self, etypes, mtime):
 """function designed to be called from an external repository which
 is using this one as a rql source for synchronization, and return a
 3-uple containing :
 * the local date
 return updatetime, modentities, delentities
 finally:
 session.close()
 # session handling ########################################################
 def close_sessions(self):
 """close every opened sessions"""
 for sessionid in self._sessions.keys():
 try:
 self.close(sessionid)
 for session in self._sessions.values():
 if session.timestamp < mintime:
 self.close(session.id)
 nbclosed += 1
 return nbclosed
 def internal_session(self, cnxprops=None):
 """return a dbapi like connection/cursor using internal user which
 have every rights on the repository. You'll *have to* commit/rollback
 or close (rollback implicitly) the session once the job's done, else
 you'll leak connections pool up to the time where no more pool is
 available, causing irremediable freeze...
 """
 session = InternalSession(self, cnxprops)
 session.set_pool()
 return session
 def _get_session(self, sessionid, setpool=False):
 """return the user associated to the given session identifier"""
 try:
 session = self._sessions[sessionid]
 except KeyError:
 # data sources handling ###################################################
 # * correspondance between eid and (type, source)
 # * correspondance between eid and local id (i.e. specific to a given source)
 # * searchable text indexes
 def type_and_source_from_eid(self, eid, session=None):
 """return a tuple (type, source, extid) for the entity with id <eid>"""
 try:
 eid = typed_eid(eid)
 except ValueError:
 except KeyError:
 etype = None
 rqlcache.pop('Any X WHERE X eid %s' % eid, None)
 for source in self.sources:
 source.clear_eid_cache(eid, etype)
 def type_from_eid(self, eid, session=None):
 """return the type of the entity with id <eid>"""
 return self.type_and_source_from_eid(eid, session)[0]
 def source_from_eid(self, eid, session=None):
 """return the source for the given entity's eid"""
 return self.sources_by_uri[self.type_and_source_from_eid(eid, session)[1]]
 def eid2extid(self, source, eid, session=None):
 """get local id from an eid"""
 etype, uri, extid = self.type_and_source_from_eid(eid, session)
 if source.uri != uri:
 # eid not from the given source
 session.commit(reset_pool)
 return eid
 except:
 session.rollback(reset_pool)
 raise
 def add_info(self, session, entity, source, extid=None, complete=True):
 """add type and source info for an eid into the system table,
 and index the entity with the full text index
 """
 # begin by inserting eid/type/source/extid into the entities table
 session.add_query_data('neweids', entity.eid)
 # now we can update the full text index
 if self.do_fti:
 FTIndexEntityOp(session, entity=entity)
 CleanupEidTypeCacheOp(session)
 def delete_info(self, session, eid):
 self._prepare_delete_info(session, eid)
 self._delete_info(session, eid)
 def _prepare_delete_info(self, session, eid):
 """prepare the repository for deletion of an entity:
 * update the fti
 * mark eid as being deleted in session info
 * setup cache update operation
 """
 self.system_source.fti_unindex_entity(session, eid)
 pending = session.query_data('pendingeids', set(), setdefault=True)
 pending.add(eid)
 CleanupEidTypeCacheOp(session)
 def _delete_info(self, session, eid):
 """delete system information on deletion of an entity:
 * delete all relations on this entity
 * transfer record from the entities table to the deleted_entities table
 """
 etype, uri, extid = self.type_and_source_from_eid(eid, session)
 self._clear_eid_relations(session, etype, eid)
 self.system_source.delete_info(session, eid, etype, uri, extid)
 def _clear_eid_relations(self, session, etype, eid):
 """when a entity is deleted, build and execute rql query to delete all
 its relations
 """
 rql = []
 if entity.eid in alreadydone:
 self.info('skipping reindexation of %s, already done', entity.eid)
 return
 alreadydone.add(entity.eid)
 self.system_source.fti_index_entity(session, entity)
 def locate_relation_source(self, session, subject, rtype, object):
 subjsource = self.source_from_eid(subject, session)
 objsource = self.source_from_eid(object, session)
 if not (subjsource is objsource and subjsource.support_relation(rtype, 1)):
 source = self.system_source
 if not source.support_relation(rtype, 1):
 raise RTypeNotSupportedBySources(rtype)
 else:
 source = subjsource
 return source
 def locate_etype_source(self, etype):
 for source in self.sources:
 if source.support_entity(etype, 1):
 return source
 else:
 raise ETypeNotSupportedBySources(etype)
 def glob_add_entity(self, session, entity):
 """add an entity to the repository
 the entity eid should originaly be None and a unique eid is assigned to
 the entity instance
 """
 entity = entity.pre_add_hook()
 eschema = entity.e_schema
 self.hm.call_hooks('before_add_relation', attr, session,
 entity.eid, attr, value)
 self.hm.call_hooks('after_add_relation', attr, session,
 entity.eid, attr, value)
 return entity.eid
 def glob_update_entity(self, session, entity):
 """replace an entity in the repository
 the type and the eid of an entity must not be changed
 """
 #print 'update', entity
 self._delete_info(session, eid)
 source.delete_entity(session, etype, eid)
 if source.should_call_hooks:
 self.hm.call_hooks('after_delete_entity', etype, session, eid)
 # don't clear cache here this is done in a hook on commit
 def glob_add_relation(self, session, subject, rtype, object):
 """add a relation to the repository"""
 assert subject is not None
 assert rtype
 assert object is not None
 self.hm.call_hooks('after_delete_relation', rtype, session,
 subject, rtype, object)
 # pyro handling ###########################################################
 def pyro_register(self, host=''):
 """register the repository as a pyro object"""
 from Pyro import core
 port = self.config['pyro-port']
 nshost, nsgroup = self.config['pyro-ns-host'], self.config['pyro-ns-group']
 daemon.connect(impl, '%s.%s' % (nsgroup, nsid))
 msg = 'repository registered as a pyro object using group %s and id %s'
 self.info(msg, nsgroup, nsid)
 self.pyro_registered = True
 return daemon
 def pyro_nameserver(self, host=None, group=None):
 """locate and bind the the name server to the daemon"""
 from Pyro import naming, errors
 # locate the name server
 nameserver = naming.NameServerLocator().getNS(host)
 except errors.NamingError:
 pass
 return nameserver
 # multi-sources planner helpers ###########################################
 @cached
 def rel_type_sources(self, rtype):
 return [source for source in self.sources
 if source.support_relation(rtype)
 or rtype in source.dont_cross_relations]
 @cached
 def can_cross_relation(self, rtype):
 return [source for source in self.sources
 if source.support_relation(rtype)
 and rtype in source.cross_relations]
 @cached
 def is_multi_sources_relation(self, rtype):
 return any(source for source in self.sources
 if not source is self.system_source
 and source.support_relation(rtype))
 def pyro_unregister(config):
 """unregister the repository from the pyro name server"""
 nshost, nsgroup = config['pyro-ns-host'], config['pyro-ns-group']
 appid = config['pyro-id'] or config.appid

branch	tls-sprint
changeset 1482	93c613913912
parent 1398	5fe84a5f7035
child 1664	03ebeccf9f1d