SQLAlchemy: Arbeiten mit mehreren QThreads?

[snafu]

Jetzt vergibst du wieder die Original-Session an verschiedene Threads. So sollte das ja eigentlich nicht sein. Ich meinte die Rückgabe von scoped_session() bloß im Rahmen eines Kontext-Managers. Wenn du diesen Kontext-Manager auf höherer Ebene letztlich nur einmal ausführst, dann hebelst du die Vorteile von scoped_session() ja wieder aus.

Die Idee ist ja, dass man von verschiedenen Stellen im Programm auf die gleiche Session zugreift ohne dass die Session explizit übergeben werden - und das auch optional über verschiedene Threads ohne dass sich die Threads in die Quere kommen. So wie du das jetzt benutzt, ist dir die Anwendung scheinbar noch nicht ganz klar geworden.

[Sophus]

@snafu: Ich bin verwirrt. Meintest du denn so: Scoped_session, anstatt so: Scoped_session()?

build_scope_session = Scoped_session --> # sqlalchemy.orm.scoping.scoped_session
build_scope_session = Scoped_session () --> # this is sqlalchemy.orm.session.Session

Ich verfolge dabei diesen Kapitel der SQLAlchemy-Dokumentation: When do I construct a Session, when do I commit it, and when do I close it?.
Dort wird eine Beispiel-Idee gezeigt.

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()

Und dieses Beispiel habe ich verfolgt: Und so wie ich das verstehe, wird der Kontext-Manager einmalig geöffnet, die Session werden an die Threads übergeben?

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)

[snafu]

Habe hier mal grob skizziert wie man scoped_session() benutzen kann:

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()

[snafu]

Und dieses Beispiel habe ich verfolgt: Und so wie ich das verstehe, wird der Kontext-Manager einmalig geöffnet, die Session werden an die Threads übergeben?

Wo siehst du da Threads? Wenn alles im selben Thread passiert, dann ist ja klar, dass die zu teilende Session ruhig auf einer höheren Ebene erstellt werden kann und dann z.B. in einer Schleife an die verarbeitenden Funktionen weitergereicht wird. scoped_session() kümmert sich laut Doku aber auch darum, dass die Threads sich bei der Verwendung der geteilten Session nicht in die Quere kommen. Vielleicht bin ich hier ja etwas übervorsichtig, aber ich würde die Rückgabe von scoped_session() in einer multi-threaded Anwendung erst so spät wie möglich aufrufen. Es tut nicht weh und vermeidet möglicherweise subtile Fehler. Muss aber jeder für sich selber wissen...

[snafu]

Na du wirst mein Beispiel sicher nicht 1 zu 1 übernommen haben. Aber wenn es so für dich funktioniert, dann ist ja alles prima.

EDIT: Hier stand vorher noch ein Kommentar von Sophus...

[Sophus]

Na du wirst mein Beispiel sicher nicht 1 zu 1 übernommen haben. Aber wenn es so für dich funktioniert, dann ist ja alles prima.

EDIT: Hier stand vorher noch ein Kommentar von Sophus...

Ich habe meinen Beitrag zurück gezogen. Mein Fehler war, dass ich den Rückgabewert von scoped_session() in den Threads zu früh aufgerufen habe und dadurch eine ellenlange Fehlermeldung bekam. Erst als ich bei dir las, dass man den Rückgabewert so spät wie möglich aufrufen soll, funktionierte deine Variante wieder. Wie gesagt, auch ohne den Rückgabewert aufzurufen hat geklappt. Das heißt, selbst mit dem Rückgabewert von scoped_session() wurden alle Daten ausgegeben, und das auch nach mehreren Male. Irgendwie scheint sowohl deine Version als auch meine Version zu klappen. Ich werde das mal beobachten

Ich danke dir vielmals.

EDIT:

P.S. Hier (What is the difference between Session and db.session in SQLAlchemy?) wird sogar (zum Thema Flask) das thematisiert, was ich mit dem Rückgabewert von scoped_session() meine. Der Antworter meint selbst, dass man mit <class 'sqlalchemy.orm.scoping.scoped_session'> arbeiten kann, und demonstriert es.

[snafu]

Aber dann mal ganz ehrlich: Wenn beides für dich aufs Gleiche hinausläuft (was es - nebenbei gesagt - ja gerade nicht tut), warum verwendest du dann den Umweg über scoped_session()? IMHO kannst du dann auch direkt die Original-Session herumreichen, denn welchen Vorteil soll dies deiner Meinung nach bringen?

Wie sieht denn dein jetziger Code aus? Und es wäre toll, wenn du nicht wieder 200 Zeilen postest, sondern einfach grob skizzierst, wie du die Session herumreichst...

[snafu]

P.S. Hier (What is the difference between Session and db.session in SQLAlchemy?) wird sogar (zum Thema Flask) das thematisiert, was ich mit dem Rückgabewert von scoped_session() meine. Der Antworter meint selbst, dass man mit <class 'sqlalchemy.orm.scoping.scoped_session'> arbeiten kann, und demonstriert es.

Das hat auch niemand bestritten. Zumal es - wieder einmal - hier nicht um eine multi-threaded Anwendung geht. Irgendwie glaube ich, du wirst diesen Punkt erst verstanden haben, wenn du auf die Nase gefallen bist.

[Sophus]

@snafu: Ok, ich werde dir mal meine grobe Skizze zeigen. Diesmal ist mein Beispiel extrem gekürzt Kommentare schreibe ich für weitere Erläuterungen in dem Pseudo-Quelltext - auf englisch. Auf diese Weise kannst du meinen Gedankengang nachvollziehen.

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()

[Sophus]

@snafu: Ich habe die zweite Version mitgebracht. Diesmal habe ich das ManagedSessionScope()-Objekt einmalig beim Start des Programm geöffnet. Die with-Anweisung bleibt in diesem Beispiel so lange am Leben, wie die Applikation läuft. Ich übergebe beispielsweise der GUI-Klasse die Klasse <class 'sqlalchemy.orm.scoping.scoped_session'>. Diese Klasse wird dann innerhalb dieser GUI-Klasse klassenweit benutzt. In der start_all_selection()-Methode verzichte ich diesmal auf die with-Anweisung, schließlich habe ich sie ja in den Start (main()-Funktion) verlegt. Ich übergebe jedem Thread die gegebene Klasse. Bei jedem Thread-Start wird auch die MasterDataManipulation()-Klasse erzeugt. In dieser Klasse wird die <class 'sqlalchemy.orm.scoping.scoped_session'> aufgerufen. Sobald die select_all()-Methode mit ihrer Arbeit fertig ist, wird ein commit() abgesetzt. Aus diesem Grund wird dann die Verbindung zum Verbindungspool zurückgegebenen und die Verbindung wird wieder freigegeben.

Sollte ich das immer noch falsch verstanden haben, dann kannst du mir ein Kinderbuch schreiben, mit Bilderchen, Sprechblasen, und mit den einfachsten Erklärungen. Denn ich bin mit meinen Latein am Ende Ich war nie gut in Latein

Zweite Version der Skizze:

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()

[snafu]

Ich denke, du machst es immer noch nicht so wie vorgesehen. Dieses ManagedSessionScope() ist schon komisch gemacht, weil es anscheinend die Original-Session und die ScopedSession zusammen ausliefert (wovon du später die "falsche" verwendest). Du packst quasi etwas in Schutzfolie, holst es aber sofort wieder aus der Folie raus und glaubst, dass der Versand jetzt sicher sei. Lass die ScopedSession doch einfach komplett weg. Das hätte bei deinem Vorgehen den selben Effekt.

[Sophus]

@snafu: Wo wird die Originale Session ausgeliefert? ManagedSessionScope() liefert beim Aufruf der __init__() einen Tuple. Einmal die Rückgabe von scoped_session() und dann die Klasse selbst. Aber die Klasse selbst verwende ich erst mal nicht. Ich habe keine Ahmung, wie du es sonst meinst. Vielleicht denke ich auch nur verquert.

[Sophus]

@snafu: Ich glaube, du bist von der falschen Namensgebung meinerseits irritiert. Also mein ManagedSessionScope()-Objekt liefert folgendes zurück: (extrem verkürzt)

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

    [...]

Du siehst, ich bekomme nicht die originale Session zurück. Warum ich mir dann auch noch self mit zurückschicken lasse? Eigentlich könnte ich sie auch weglassen. Aber verwenden tue ich die self nicht. Ich glaube, das hat dich ein wenig verwirrt. War mein Fehler.

Gedankenproblem:
Aber ich habe ein kleines Gedankenproblem. In meiner zurvor vorgestellten Skizze wird ManagedSessionScope() durch die with-Anweisung direkt am Anfang des Programmstart geöffnet. Dadurch habe ich dann eine session, die ich applikationsweit benutzen könnte. Aber das war ja nur ein Beispiel. In meinem richtigen Projekt ist es so, dass sich der Benutzer später in die Datenbank anmelden kann - wann immer er/sie will. Folgende Situation, der Benutzer startet das Programm, geht zu einem späteren Zeitpunkt in die Menüleiste, und geht auf "Anmelden", ein neues Fenster, geerbt von QDialog(), wird erzeugt. Benutzer gibt seine Anmelde-Daten ein, klickt dann auf die "Ok"-Schaltfläche. Anmeldung läuft durch. Wenn alles gut ging, bekommt der Benutzer eine MessageBox angezeigt, mit der Information, dass alles wunderbar verlaufen ist, ansonsten werden entsprechende Fehlermeldungen über die MessageBox ausgegeben. Sobald der Benutzer diese MessageBox, mit der erfolgreichen Meldung, gesehen hat, schließt sich das Anmelde-Fenster. Und genau JETZT habe ich ein Problem. Wo halte ich die Session während dessen am "leben"? Anders gefragt: wie/wo soll ich die with-Anweisung die ganze Zeit geöffnet lassen, damit die Applikation auch nur mit einer gültigen Session arbeitet? Ich meine, in meiner vorherigen Skizze war die with-Anweisung durch sys.exit(app.exec_()) offen geblieben, denn dies wirkte fast wie eine while-Schleife. Meine erste Überlegung ist, dass ich einen weiteren Thread konstruiere, der nur dafür da ist, die with-Anweisung offen zu halten, damit wir eine gültige Session applikationsweis haben? In diesem besagten Thread müsste ich dann einen QTimer() einbauen, diesen Timer dann in den Korpus der with-Anweisung packen. Ansonsten fällt mir keine Idee ein, wie ich eine with-Anweisung zum späteren Zeitpunkt ununterbrochen(solange wie die Applikation läuft) geöffnet halten soll. Sonst hätte ich ja den gleichen Effekt, die ich bisher gemacht habe, dass jedesmal, bei Gebrauch erneut die with-Anweisung durchgeführt wurde. Würde ich dies tun, hätte ich jedesmal eine andere Session, und keine einzig gültige Session. Ich stecke gerade gedanklich fest.

[BlackJack]

@Sophus: Du gibst bei `__enter__()` ein Tupel zurück? Aber das hat doch dann gar keine `__exit__()`-Methode!

[Sophus]

Ich werde den Tuple wieder entfernen. Ihr habt Recht, es war eine Dumme Idee Ich gebe nur noch die scoped_session zurück.

[snafu]

Die Frage ist, warum du überhaupt die scoped_session nochmals wrappst, denn das ist ja schon ein Wrapper. Wäre es nicht einfacher, eine "normale" Funktion zu programmieren, die eine vorkonfigurierte Sitzung als scoped_session liefert?

Im Übrigen ist es keine so gute Idee, eine Session während des gesamten Programmlaufs offen zu halten. Du solltest besser die Nutzereingaben aus der GUI sammeln und wenn er auf "Übermitteln" oder sowas klickt, dann wird die Session geöffnet und mit den Daten gefüttert, die commited werden sollen. Danach schließt man sie wieder. Und während dieses Ablaufs kann die offene Session (ScopedSession) dann halt auch - abhängig von deiner Programmstruktur - herumgereicht werden.

[Sophus]

@snafu: Aber dann lande ich ja bei dem Ausgangspunkt, den du kritisiert hast. Ursprünglich sah die Struktur wie folgt aus, in der GUI-Klasse (MyCustomDialog()) gab es eine start_all_selection()-Methode, die so konstruiert war, dass bei jedem Klick auf die Schaltfläche die ManagedSessionScope() geöffnet wurde, die Rückgabe von scoped_session() an die Trheads weiterreichte, und nach getaner Arbeit wurde die ManagedSessionScope() automatisch wieder beendet. Und das hast du unter anderem doch kritisiert?

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)

Daher war meine Idee, einen Thread zu konstruieren, der die Session solange offen hält, ehe der Benutzer in der Menüleiste auf "Abmelden" klickt, und der laufende Thread beendet wird. Im Thread wäre dann eine while-Schleife, die dann vor dem Beenden des Threads unterbrochen wird, so dass man den Korpus der with-Anweisung verlässt. Klickt also der Anwender auf "Abmelden", wird die while-Schleife im Thread beendet und der Thread wird beendet, und die Session ist geschlossen. Beim nächsten Mal, wenn er sich entschließt, sich anzumelden, wird wieder ein neuer Thread erzeugt, eine neue Session eröffnet und so weiter und so fort.

Ungefähr so: Die Signale habe ich jetzt mal der Übersichthalber ausgelassen, wollte jetzt nicht den vollständigen Quelltext kopieren. Aber so dachte ich mir, könnte das aussehen?

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

[__deets__]

Was hat denn eine Session bei *Bedarf* zu oeffnen damit zu tun, das die dann ploetzlich (und fehlerhafterweise) von allen moeglichen Threads geteilt werden muss? Dieser Schluss ist Unsinn.

Jeder Thread braucht seine eigene Session, aber die muss er bei Bedarf oeffnen, und dann wieder schliessen, wenn er sie nicht braucht. Und das ist nicht ueber seine ganze Lebenszeit, sondern waehrend einer bestimmten semantisch sinnvollen Aktion, zB dem auslesen von Werten fuer die Dropdowns, oder dem eintragen eines neuen Datensatzes. Kann also durchaus eine handvoll Statements beinhalten, aber eben nicht ueber die ganze Lebenszeit des Programms.

[Sophus]

@__deets__: Wenn ich dich richtig verstehe, für jeden Thread eine with-Anweisung konstruieren, und nicht so wie ich es bisher gemacht habe, eine with-Anweisung für alle Threads, die dann am Ende wieder geschlossen wird, wenn der letzte Thread fertig ist? Das heißt in meinem Beispiel, dass ich ganze 8 Twith-Anweisungen anwende?

[__deets__]

Ja, wobei du das komisch ausdrueckst. Je nach Programmablauf und Code-Struktur kann das auch nur *eine* with-Anweisung sein, aber die wird halt oft aufgerufen. Oder es sind 2000, weil du eine um jeden SQL-code-block schreibst. Aber das entscheidende ist: kein Thread teilt seine Session mit einem anderen, das fuehrt zur Katastrophe.