Das deutsche Python-Forum

Code: Alles auswählen

### this is the **wrong way to do it** ###

class ThingOne(object):
    def go(self):
        session = Session()
        try:
            session.query(FooBar).update({"x": 5})
            session.commit()
        except:
            session.rollback()
            raise

class ThingTwo(object):
    def go(self):
        session = Session()
        try:
            session.query(Widget).update({"q": 18})
            session.commit()
        except:
            session.rollback()
            raise

def run_my_program():
    ThingOne().go()
    ThingTwo().go()

Code: Alles auswählen

### another way (but again *not the only way*) to do it ###

from contextlib import contextmanager

@contextmanager
def session_scope():
    """Provide a transactional scope around a series of operations."""
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()


def run_my_program():
    with session_scope() as session:
        ThingOne().go(session)
        ThingTwo().go(session)

Code: Alles auswählen

def do_work(sessiongetter, value):
    shared_session = sessiongetter()
    try:
        # do stuff
    except Exception:
        shared_session.rollback()
        raise
    # NOTE: do not close the shared session!


def main():
    # ...
    engine = create_engine(dburi,
                           pool_size=10,
                           max_overflow=10,
                           pool_timeout=60,
                           echo=echo_verbose)
    sessiongetter = scoped_session(sessionmaker(bind=engine))
    try:
        for bla in blupp:
            do_work(sessiongetter, bla)
    finally:
        sessiongetter.close()

Code: Alles auswählen

class MyCustomDialog(QDialog):
 
    finish = pyqtSignal()
 
    def __init__(self, url=None, parent=None):
        QDialog.__init__(self, parent)

        self._url = url

        [...]

    def start_all_selection(self):
        # User clicks on button, which is connected with this method.
        # He wants to see all data.
        # let us start all threads - currently there a 8 threads.

        try:
            # We know that our ManagedSessionScope()-object is a custom context manager with sperate class.
            # Our custom context manager returns <class 'sqlalchemy.orm.scoping.scoped_session'>,
            # no normal session.  In this case, we use one session per thread, I think. 
            # We don't want to share nothing between threads, right?
            # Well, the user wants the program to query all data. lets do it.
            with ManagedSessionScope(url=self._url, echo_verbose=False) as (session, ScopeSession):

                [...]
                for category, combobox in list_tuple:
                    self.start_thread(combo_box=combobox,
                                      session=session,
                                      time_interval=100,
                                      category=category)

                # After all 8 threads have completed tasks and finished, 
                # the created session will close automatically. We don't need the created session anymore.
                # When the user wants the program to load the data again, we can create a new session again.

        except SQLAlchemyError as err:
            # do stuff with raised exception

        except OperationalError as OpErr:
            # du stuff with raised exception

def main():
    # Image, somewhere in the program there is a login window
    # where the user can enter his login data. In this example,
    # the user enters his login data at start start of the program.
    dbms = raw_input('Enter database type: ')
    dbdriver = raw_input('Enter database driver: ')
    dbuser = raw_input('Enter user name: ')
    dbuser_pwd = raw_input('Enter user password: ')
    db_server_host = raw_input('Enter server host: ')
    dbport = raw_input('Enter port: ')
    db_name = raw_input('Enter database name: ')

    url = '{}+{}://{}:{}@{}:{}/{}'.format(
           dbms, dbdriver, dbuser, dbuser_pwd, db_server_host, dbport, db_name)

    # Now we have the required url for the login. We pass
    # the url as an argument to the gui window

    [...]  

    app = QApplication(sys.argv)
    window = MyCustomDialog(url = url)
    [...]
        
if __name__ == "__main__":
    main()

Code: Alles auswählen

class MasterDataManipulation(object):

    def __init__(self, session_object=None):
    	# This class  is created on each thread and communicates with the database.
    	#self._session = session_object
    	self._session = session_object()
                 
    def select_all(self, category):

        dict_store_session_query = {'person_gender':               lambda: self._session.query(PERSON_GENDER),
                                     'person_nationality':          lambda: self._session.query(PERSON_NATIONALITY),}

        try:

            for record in dict_store_session_query[category]():
                if category == 'person_gender':
                    yield record.id, record.gender
                if category == 'person_nationality':

            # I use commit() on given session for flushing the connection.
            # That means, the connetion is flushed, the querys
            # are committed, the connection object closed
            # and discarded, the underlying DBAPI connection
            # returned to the connection pool.
            self._session.commit()

        except Exception:
            self._session.rollback()

class MyCustomDialog(QDialog):
 
    finish = pyqtSignal()
 
    def __init__(self, session=None, parent=None):
        QDialog.__init__(self, parent)
        # Now we save <class 'sqlalchemy.orm.scoping.scoped_session'> in the attribute named 'self._session'
        self._session = session
       [...]

    def start_all_selection(self):
        [...]
        try:
                
            for category, combobox in list_tuple:

                combobox.clear()

                self.start_thread(combo_box=combobox,
                                  session=self._session,
                                  time_interval=100,
                                  category=category)

        except SQLAlchemyError as err:
            # Do stuff with this raised exception

def main():
    # Image, somewhere in the program there is a Log_In window
    # where the user can enter his LoGi data. In this example,
    # the user enters his Login data at start start of the program.
    # The user can LogIn to the database-server 
    dbms = raw_input('Enter database type: ')
    dbdriver = raw_input('Enter database driver: ')
    dbuser = raw_input('Enter user name: ')
    dbuser_pwd = raw_input('Enter user password: ')
    db_server_host = raw_input('Enter server host: ')
    dbport = raw_input('Enter port: ')
    db_name = raw_input('Enter database name: ')

    url = '{}+{}://{}:{}@{}:{}/{}'.format(
           dbms, dbdriver, dbuser, dbuser_pwd, db_server_host, dbport, db_name)

    # Now we have the required url for the login. We pass
    # the url as an argument to the gui window

    try:
        with ManagedSessionScope(url=url, echo_verbose=True) as (session, ScopeSession):
            
            app = QApplication(sys.argv)
            window = MyCustomDialog(session = session)
            window.show()
            sys.exit(app.exec_())
            
    except TypeError:
        
        print "ERROR", format_exc(exc_info())
        
if __name__ == "__main__":
    main()

Code: Alles auswählen

class ManagedSessionScope(object):
    def __init__(self, url=None,
                 echo_verbose=True,):
        [...]
        self._Engine = create_engine(self.url, echo=self.echo_verbose) 

        self._Session = scoped_session(sessionmaker(bind=self._Engine))

    def __enter__(self):
        self.session = self._Session     # this is now a scoped session
                                                      # sqlalchemy.orm.scoping.scoped_session
        return self.session, self

    [...]

Code: Alles auswählen

class MyCustomDialog(QDialog):
 
    finish = pyqtSignal()
 
    def __init__(self, scoped_session=None, parent=None):
        QDialog.__init__(self, parent)

        [...]

    def start_all_selection(self):
    [...]
        with ManagedSessionScope(url=self._url) as session_that_scoped:
            
            for category, combobox in list_tuple:
                combobox.clear()
                
                self.start_thread(combo_box=combobox,session=session_that_scoped, category=category)

Code: Alles auswählen

class SessionTask(QObject):
 
    def __init__(self, parent=None):
        QObject.__init__(self, parent)
        self.stop_loop = False
 
    def run(self):
        with ManagedSessionScope(url=self._url) as session:
            # do stuff with given scoped_session()

            while not self.stop_loop:  pass           
 
    def stop(self):
        self.stop_loop = True