Source code for destral.transaction

from threading import local

from destral.openerp import OpenERPService

[docs]class Singleton(type): """Metaclass for singleton pattern. :copyright: Tryton Project """ def __init__(mcs, name, bases, dict_): super(Singleton, mcs).__init__(name, bases, dict_) mcs.instance = None def __call__(mcs, *args, **kwargs): if mcs.instance is None: mcs.instance = super(Singleton, mcs).__call__(*args, **kwargs) return mcs.instance
[docs]class Transaction(local): """Transaction object """ __metaclass__ = Singleton database = None service = None pool = None cursor = None user = None context = None def __init__(self): pass
[docs] def start(self, database_name, user=1, context=None): """Start a new transaction :param database_name: Database name :param user: User id :param context: Context to be used """ self._assert_stopped() self.service = OpenERPService(db_name=database_name) self.pool = self.service.pool self.cursor = self.service.db.cursor() self.user = user self.context = context if context is not None else self.get_context() return self
[docs] def stop(self): """Stop the transaction. """ self.cursor.close() self.service = None self.cursor = None self.user = None self.context = None self.database = None self.pool = None
[docs] def get_context(self): """Loads the context of the current user """ assert self.user is not None user_obj = self.pool.get('res.users') return user_obj.context_get(self.cursor, self.user)
def __enter__(self): return self def __exit__(self, type, value, traceback): self.stop() def _assert_stopped(self): assert self.service is None assert self.database is None assert self.cursor is None assert self.pool is None assert self.user is None assert self.context is None