cubicweb: comparison server/repository.py

equal deleted inserted replaced

-:8af7c6d86efb
+:a964c40adbe3
-# copyright 2003-2011 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
+# copyright 2003-2012 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
 #
 # This file is part of CubicWeb.
 #
 # CubicWeb is free software: you can redistribute it and/or modify it under the
 NO_CACHE_RELATIONS = set( [('owned_by', 'object'),
 ('created_by', 'object'),
 ('cw_source', 'object'),
 ])
-def prefill_entity_caches(entity, relations):
+def prefill_entity_caches(entity):
 session = entity._cw
 # prefill entity relation caches
 for rschema in entity.e_schema.subject_relations():
 rtype = str(rschema)
 if rtype in schema.VIRTUAL_RTYPES or (rtype, 'subject') in NO_CACHE_RELATIONS:
 session.execute('DELETE X %s Y WHERE Y eid %%(y)s, '
 'NOT X eid %%(x)s' % rtype,
 {'x': eidfrom, 'y': eidto})
+class NullEventBus(object):
+def publish(self, msg):
+pass
+def add_subscription(self, topic, callback):
+pass
+def start(self):
+pass
+def stop(self):
+pass
 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):
+def __init__(self, config, tasks_manager=None, vreg=None):
 self.config = config
 if vreg is None:
-vreg = cwvreg.CubicWebVRegistry(config)
+vreg = cwvreg.CWRegistryStore(config)
 self.vreg = vreg
+self._tasks_manager = tasks_manager
 self.pyro_registered = False
 self.pyro_uri = None
+self.app_instances_bus = NullEventBus()
 self.info('starting repository from %s', self.config.apphome)
 # dictionary of opened sessions
 self._sessions = {}
 # list of functions to be called at regular interval
-self._looping_tasks = []
 # list of running threads
 self._running_threads = []
 # initial schema, should be build or replaced latter
 self.schema = schema.CubicWebSchema(config.appid)
 self.vreg.schema = self.schema # until actual schema is loaded...
 # open some connections set
 if config.init_cnxset_pool:
 self.init_cnxset_pool()
 @onevent('after-registry-reload', self)
 def fix_user_classes(self):
+# After registery reload the 'CWUser' class used for CWEtype
+# changed.  To any existing user object have a different class than
+# the new loaded one. We are hot fixing this.
 usercls = self.vreg['etypes'].etype_class('CWUser')
 for session in self._sessions.values():
 if not isinstance(session.user, InternalManager):
 session.user.__class__ = usercls
 self.sources_by_eid = {}
 if self.config.quick_start \
 or not 'CWSource' in self.schema: # # 3.10 migration
 self.system_source.init_creating()
 return
-session = self.internal_session()
+with self.internal_session() as session:
-try:
 # FIXME: sources should be ordered (add_entity priority)
 for sourceent in session.execute(
 'Any S, SN, SA, SC WHERE S is_instance_of CWSource, '
 'S name SN, S type SA, S config SC').entities():
 if sourceent.name == 'system':
 self.system_source.eid = sourceent.eid
 self.sources_by_eid[sourceent.eid] = self.system_source
 self.system_source.init(True, sourceent)
 continue
 self.add_source(sourceent, add_to_cnxsets=False)
-finally:
-session.close()
 def _clear_planning_caches(self):
 for cache in ('source_defs', 'is_multi_sources_relation',
 'can_cross_relation', 'rel_type_sources'):
 clear_cache(self, cache)
 for source in self.sources_by_uri.values():
 source.set_schema(schema)
 self.schema = schema
 def fill_schema(self):
-"""lod schema from the repository"""
+"""load schema from the repository"""
 from cubicweb.server.schemaserial import deserialize_schema
 self.info('loading schema from the repository')
 appschema = schema.CubicWebSchema(self.config.appid)
 self.set_schema(self.config.load_bootstrap_schema(), resetvreg=False)
 self.debug('deserializing db schema into %s %#x', appschema.name, id(appschema))
-session = self.internal_session()
+with self.internal_session() as session:
-try:
 try:
 deserialize_schema(appschema, session)
 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]
-finally:
-session.close()
 self.set_schema(appschema)
-def start_looping_tasks(self):
+def _prepare_startup(self):
+"""Prepare "Repository as a server" for startup.
+* trigger server startup hook,
+* register session clean up task.
+"""
 if not (self.config.creating or self.config.repairing
 or self.config.quick_start):
 # call instance level initialisation hooks
 self.hm.call_hooks('server_startup', repo=self)
 # register a task to cleanup expired session
 self.cleanup_session_time = self.config['cleanup-session-time'] or 60 * 60 * 24
 assert self.cleanup_session_time > 0
 cleanup_session_interval = min(60*60, self.cleanup_session_time / 3)
-self.looping_task(cleanup_session_interval, self.clean_sessions)
+assert self._tasks_manager is not None, "This Repository is not intended to be used as a server"
-assert isinstance(self._looping_tasks, list), 'already started'
+self._tasks_manager.add_looping_task(cleanup_session_interval,
-for i, (interval, func, args) in enumerate(self._looping_tasks):
+self.clean_sessions)
-self._looping_tasks[i] = task = utils.LoopTask(self, interval, func, args)
-self.info('starting task %s with interval %.2fs', task.name,
+def start_looping_tasks(self):
-interval)
+"""Actual "Repository as a server" startup.
-task.start()
-# ensure no tasks will be further added
+* trigger server startup hook,
-self._looping_tasks = tuple(self._looping_tasks)
+* register session clean up task,
+* start all tasks.
+XXX Other startup related stuffs are done elsewhere. In Repository
+XXX __init__ or in external codes (various server managers).
+"""
+self._prepare_startup()
+assert self._tasks_manager is not None, "This Repository is not intended to be used as a server"
+self._tasks_manager.start()
 def looping_task(self, interval, func, *args):
 """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:
+assert self._tasks_manager is not None, "This Repository is not intended to be used as a server"
-self._looping_tasks.append( (interval, func, args) )
+self._tasks_manager.add_looping_task(interval, func, *args)
-except AttributeError:
-raise RuntimeError("can't add looping task once the repository is started")
 def threaded_task(self, func):
 """start function in a separated thread"""
-t = utils.RepoThread(func, self._running_threads)
+utils.RepoThread(func, self._running_threads).start()
-t.start()
 #@locked
 def _get_cnxset(self):
 try:
 return self._cnxsets_pool.get(True, timeout=5)
 def shutdown(self):
 """called on server stop event to properly close opened sessions and
 connections
 """
 assert not self.shutting_down, 'already shutting down'
+if not (self.config.creating or self.config.repairing
+or self.config.quick_start):
+# then, the system source is still available
+self.hm.call_hooks('before_server_shutdown', repo=self)
 self.shutting_down = True
 self.system_source.shutdown()
-if isinstance(self._looping_tasks, tuple): # if tasks have been started
+if self._tasks_manager is not None:
-for looptask in self._looping_tasks:
+self._tasks_manager.stop()
-self.info('canceling task %s...', looptask.name)
+if not (self.config.creating or self.config.repairing
-looptask.cancel()
+or self.config.quick_start):
-looptask.join()
+self.hm.call_hooks('server_shutdown', repo=self)
-self.info('task %s finished', looptask.name)
 for thread in self._running_threads:
 self.info('waiting thread %s...', thread.getName())
 thread.join()
 self.info('thread %s finished', thread.getName())
-if not (self.config.creating or self.config.repairing
-or self.config.quick_start):
-self.hm.call_hooks('server_shutdown', repo=self)
 self.close_sessions()
 while not self._cnxsets_pool.empty():
 cnxset = self._cnxsets_pool.get_nowait()
 try:
 cnxset.close(True)
 def check_auth_info(self, session, login, authinfo):
 """validate authentication, raise AuthenticationError on failure, return
 associated CWUser's eid on success.
 """
-for source in self.sources:
+# iter on sources_by_uri then check enabled source since sources doesn't
-if source.support_entity('CWUser'):
+# contain copy based sources
+for source in self.sources_by_uri.itervalues():
+if self.config.source_enabled(source) and source.support_entity('CWUser'):
 try:
 return source.authenticate(session, login, **authinfo)
 except AuthenticationError:
 continue
 else:
 results['type_source_cache_size'] = len(self._type_source_cache)
 results['extid_cache_size'] = len(self._extid_cache)
 results['sql_no_cache'] = self.system_source.no_cache
 results['nb_open_sessions'] = len(self._sessions)
 results['nb_active_threads'] = threading.activeCount()
-results['looping_tasks'] = ', '.join(str(t) for t in self._looping_tasks)
+looping_tasks = self._tasks_manager._looping_tasks
+results['looping_tasks'] = ', '.join(str(t) for t in looping_tasks)
 results['available_cnxsets'] = self._cnxsets_pool.qsize()
 results['threads'] = ', '.join(sorted(str(t) for t in threading.enumerate()))
 return results
 def gc_stats(self, nmax=20):
 This is a public method, not requiring a session id.
 """
 from logilab.common.changelog import Version
 vcconf = {}
-session = self.internal_session()
+with self.internal_session() as session:
-try:
 for pk, version in session.execute(
 'Any K,V WHERE P is CWProperty, P value V, P pkey K, '
 'P pkey ~="system.version.%"', build_descr=False):
 cube = pk.split('.')[-1]
 # XXX cubicweb migration
 fsversion = self.config.cubicweb_version()
 if version < fsversion:
 msg = ('instance has %s version %s but %s '
 'is installed. Run "cubicweb-ctl upgrade".')
 raise ExecutionError(msg % (cube, version, fsversion))
-finally:
-session.close()
 return vcconf
 @cached
 def source_defs(self):
 """Return the a dictionary containing source uris as value and a
 def properties(self):
 """Return a result set containing system wide properties.
 This is a public method, not requiring a session id.
 """
-session = self.internal_session()
+with self.internal_session() as session:
-try:
 # don't use session.execute, we don't want rset.req set
 return self.querier.execute(session, 'Any K,V WHERE P is CWProperty,'
 'P pkey K, P value V, NOT P for_user U',
 build_descr=False)
-finally:
-session.close()
 # XXX protect this method: anonymous should be allowed and registration
 # plugged
 def register_user(self, login, password, email=None, **kwargs):
 """check a user with the given login exists, if not create it with the
 given password. This method is designed to be used for anonymous
 registration on public web site.
 """
-session = self.internal_session()
+with self.internal_session() as session:
 # for consistency, keep same error as unique check hook (although not required)
 errmsg = session._('the value "%s" is already used, use another one')
-try:
 if (session.execute('CWUser X WHERE X login %(login)s', {'login': login},
 build_descr=False)
 or session.execute('CWUser X WHERE X use_email C, C address %(login)s',
 {'login': login}, build_descr=False)):
 qname = role_name('login', 'subject')
 raise ValidationError(None, {qname: errmsg % d['email']})
 session.execute('INSERT EmailAddress X: X address %(email)s, '
 'U primary_email X, U use_email X '
 'WHERE U login %(login)s', d, build_descr=False)
 session.commit()
-finally:
-session.close()
 return True
 def find_users(self, fetch_attrs, **query_attrs):
 """yield user attributes for cwusers matching the given query_attrs
 (the result set cannot survive this method call)
 if not rschema.meta)
 cwuserattrs = self._cwuser_attrs
 for k in chain(fetch_attrs, query_attrs.iterkeys()):
 if k not in cwuserattrs:
 raise Exception('bad input for find_user')
-session = self.internal_session()
+with self.internal_session() as session:
-try:
 varmaker = rqlvar_maker()
 vars = [(attr, varmaker.next()) for attr in fetch_attrs]
 rql = 'Any %s WHERE X is CWUser, ' % ','.join(var[1] for var in vars)
 rql += ','.join('X %s %s' % (var[0], var[1]) for var in vars) + ','
 rset = session.execute(rql + ','.join('X %s %%(%s)s' % (attr, attr)
 for attr in query_attrs.iterkeys()),
 query_attrs)
 return rset.rows
-finally:
-session.close()
 def connect(self, login, **kwargs):
 """open a connection for a given user
 base_url may be needed to send mails
 raise `AuthenticationError` if the authentication failed
 raise `ConnectionError` if we can't open a connection
 """
 # use an internal connection
-session = self.internal_session()
+with self.internal_session() as session:
 # try to get a user object
 cnxprops = kwargs.pop('cnxprops', None)
-try:
 user = self.authenticate_user(session, login, **kwargs)
-finally:
-session.close()
 session = Session(user, self, cnxprops)
 user._cw = user.cw_rset.req = session
 user.cw_clear_relation_cache()
 self._sessions[session.id] = session
 self.info('opened session %s for user %s', session.id, login)
 session.commit()
 session.close()
 del self._sessions[sessionid]
 self.info('closed session %s for user %s', sessionid, session.user.login)
+def call_service(self, sessionid, regid, async, **kwargs):
+"""
+See :class:`cubicweb.dbapi.Connection.call_service`
+and :class:`cubicweb.server.Service`
+"""
+def task():
+session = self._get_session(sessionid, setcnxset=True)
+service = session.vreg['services'].select(regid, session, **kwargs)
+try:
+return service.call(**kwargs)
+finally:
+session.rollback() # free cnxset
+if async:
+self.info('calling service %s asynchronously', regid)
+self.threaded_task(task)
+else:
+self.info('calling service %s synchronously', regid)
+return task()
 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)
 modified since the given timestamp (actually entities whose full text
 index content has changed)
 * list of (etype, eid) of entities of the given types which have been
 deleted since the given timestamp
 """
-session = self.internal_session()
+with self.internal_session() as session:
 updatetime = datetime.utcnow()
-try:
 modentities, delentities = self.system_source.modified_entities(
 session, etypes, mtime)
 return updatetime, modentities, delentities
-finally:
-session.close()
 # session handling ########################################################
 def close_sessions(self):
 """close every opened sessions"""
 entity.eid = self.system_source.create_eid(session)
 # set caches asap
 extid = self.init_entity_caches(session, entity, source)
 if server.DEBUG & server.DBG_REPO:
 print 'ADD entity', self, entity.__regid__, entity.eid, edited
-relations = []
+prefill_entity_caches(entity)
-prefill_entity_caches(entity, relations)
 if source.should_call_hooks:
 self.hm.call_hooks('before_add_entity', session, entity=entity)
-activintegrity = session.is_hook_category_activated('activeintegrity')
+relations = []
+activeintegrity = session.is_hook_category_activated('activeintegrity')
 for attr in edited.iterkeys():
 rschema = eschema.subjrels[attr]
 if not rschema.final: # inlined relation
 value = edited[attr]
 relations.append((attr, value))
 session.update_rel_cache_add(entity.eid, attr, value)
 rdef = session.rtype_eids_rdef(attr, entity.eid, value)
-if rdef.cardinality[1] in '1?' and activintegrity:
+if rdef.cardinality[1] in '1?' and activeintegrity:
 with security_enabled(session, read=False):
 session.execute('DELETE X %s Y WHERE Y eid %%(y)s' % attr,
 {'x': entity.eid, 'y': value})
 edited.set_defaults()
 if session.is_hook_category_activated('integrity'):
 # these are overridden by set_log_methods below
 # only defining here to prevent pylint from complaining
 info = warning = error = critical = exception = debug = lambda msg,*a,**kw: None
 def pyro_unregister(config):
 """unregister the repository from the pyro name server"""
 from logilab.common.pyro_ext import ns_unregister
 appid = config['pyro-instance-id'] or config.appid
 ns_unregister(appid, config['pyro-ns-group'], config['pyro-ns-host'])

branch	stable
changeset 8463	a964c40adbe3
parent 8456	c912d82f2166
child 8537	e30d0a7f0087
child 8547	f23ac525ddd1