cubicweb: comparison server/pool.py

equal deleted inserted replaced

-:672acc730ce5
+:d628defebc17
 """CubicWeb server connections pool :
 * the rql repository has a limited number of connections pools, each of them
 dealing with a set of connections on each source used by the repository
 * operation may be registered by hooks during a transaction, which will  be
 fired when the pool is commited or rollbacked
 This module defined the `ConnectionsPool` class and a set of abstract classes
 for operation.
 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr
 """
 __docformat__ = "restructuredtext en"
 import sys
 class ConnectionsPool(object):
 """handle connections on a set of sources, at some point associated to a
 user session
 """
 # would need a two phases commit or like, but I don't know how to do
 # this using the db-api...
 for source, cnx in self.source_cnxs.values():
 # let exception propagates
 cnx.commit()
 def rollback(self):
 """rollback the current transaction for this user"""
 for source, cnx in self.source_cnxs.values():
 # catch exceptions, rollback other sources anyway
 try:
 for _, cnx in self.source_cnxs.values():
 try:
 cnx.close()
 except:
 continue
 # internals ###############################################################
 def pool_set(self, session):
 """pool is being set"""
 self.check_connections()
 def pool_reset(self, session):
 """pool is being reseted"""
 for source, cnx in self.source_cnxs.values():
 source.pool_reset(cnx)
 def __getitem__(self, uri):
 """subscription notation provide access to sources'cursors"""
 try:
 cursor = self._cursors[uri]
 except KeyError:
 cursor = self.source_cnxs[uri][1].cursor()
 if cursor is not None:
 # None possible on sources without cursor support such as ldap
 self._cursors[uri] = cursor
 return cursor
 def sources(self):
 """return the source objects handled by this pool"""
 # implementation details of flying insert requires the system source
 # first
 yield self.source_cnxs['system']
 for uri, (source, cursor) in self.source_cnxs.items():
 if uri == 'system':
 continue
 yield source
 #return [source_cnx[0] for source_cnx in self.source_cnxs.values()]
 def source(self, uid):
 """return the source object with the given uri"""
 return self.source_cnxs[uid][0]
 def connection(self, uid):
 """return the connection on the source object with the given uri"""
 return self.source_cnxs[uid][1]
 def reconnect(self, source):
 """reopen a connection for this source"""
 source.info('trying to reconnect')
 self.source_cnxs[source.uri] = (source, source.get_connection())
 del self._cursors[source.uri]
 def check_connections(self):
 for source, cnx in self.source_cnxs.itervalues():
 newcnx = source.check_connection(cnx)
 the pool is preparing to commit. You shouldn't do anything things which
 has to be reverted if the commit fail at this point, but you can freely
 do any heavy computation or raise an exception if the commit can't go.
 You can add some new operation during this phase but their precommit
 event won't be triggered
 commit:
 the pool is preparing to commit. You should avoid to do to expensive
 stuff or something that may cause an exception in this event
 revertcommit:
 if an operation failed while commited, this event is triggered for
 all operations which had their commit event already to let them
 revert things (including the operation which made fail the commit)
 commit are rollbacked
 order of operations may be important, and is controlled according to:
 * operation's class
 """
 def __init__(self, session, **kwargs):
 self.session = session
 self.user = session.user
 self.repo = session.repo
 self.schema = session.repo.schema
 self.__dict__.update(kwargs)
 self.register(session)
 # execution information
 self.processed = None # 'precommit', 'commit'
 self.failed = False
 def register(self, session):
 session.add_operation(self, self.insert_index())
 def insert_index(self):
 """return the index of  the lastest instance which is not a
 LateOperation instance
 """
 for i, op in enumerate(self.session.pending_operations):
 if isinstance(op, (LateOperation, SingleLastOperation)):
 return i
 return None
 def handle_event(self, event):
 """delegate event handling to the opertaion"""
 getattr(self, event)()
 def precommit_event(self):
 """the observed connections pool is preparing a commit"""
 def revertprecommit_event(self):
 """an error went when pre-commiting this operation or a later one
 should revert pre-commit's changes but take care, they may have not
 been all considered if it's this operation which failed
 """
 def commit_event(self):
 """the observed connections pool is commiting"""
 raise NotImplementedError()
 def revertcommit_event(self):
 """an error went when commiting this operation or a later one
 should revert commit's changes but take care, they may have not
 been all considered if it's this operation which failed
 """
 def rollback_event(self):
 """the observed connections pool has been rollbacked
 do nothing by default, the operation will just be removed from the pool
 operation list
 """
 class LateOperation(Operation):
 """special operation which should be called after all possible (ie non late)
 operations
 """
 def insert_index(self):
 """return the index of  the lastest instance which is not a
 SingleLastOperation instance
 """
 for i, op in enumerate(self.session.pending_operations):
 return i
 return None
 class SingleOperation(Operation):
 """special operation which should be called once"""
 def register(self, session):
 """override register to handle cases where this operation has already
 been added
 """
 operations = session.pending_operations
 equivalent = operations.pop(index)
 else:
 equivalent = None
 session.add_operation(self, self.insert_index())
 return equivalent
 def equivalent_index(self, operations):
 """return the index of the equivalent operation if any"""
 equivalents = [i for i, op in enumerate(operations)
 if op.__class__ is self.__class__]
 if equivalents:
 class SingleLastOperation(SingleOperation):
 """special operation which should be called once and after all other
 operations
 """
 def insert_index(self):
 return None
 from logging import getLogger
 from cubicweb import set_log_methods

branch	tls-sprint
changeset 1802	d628defebc17
parent 1132	96752791c2b6
child 1977	606923dff11b