2021 Hinweis: Die ursprüngliche Antwort stammt aus dem Jahr 2010. Nun scheint der bessere Ansatz, wie in den Kommentaren ausgeführt, zu sein pool_recycle-Parameter .
Originalantwort von 2010 folgt.
Siehe BEARBEITEN unten für getestete Lösung
Ich habe es nicht versucht, aber vielleicht mit PoolListener ist ein Weg zu gehen?
Sie könnten etwa so vorgehen:
class MyListener(sqlalchemy.interfaces.PoolListener):
def __init__(self):
self.retried = False
def checkout(self, dbapi_con, con_record, con_proxy):
try:
dbapi_con.info() # is there any better way to simply check if connection to mysql is alive?
except sqlalchemy.exc.OperationalError:
if self.retried:
self.retried = False
raise # we do nothing
self.retried = True
raise sqlalchemy.exc.DisconnectionError
# next, code according to documentation linked above follows
e = create_engine("url://", listeners=[MyListener()])
Auf diese Weise testen wir jedes Mal, wenn die Verbindung aus dem Pool ausgecheckt werden soll, ob sie tatsächlich mit dem Server verbunden ist. Wenn nicht, geben wir sqlalchemy eine Chance, die Verbindung wiederherzustellen. Wenn das Problem danach immer noch besteht, lassen wir es los.
PS:Ich habe nicht getestet, ob das funktioniert.
Bearbeiten:Wie bei den Pylons müssten Änderungen an der oben gezeigten Engine-Initialisierung in your_app.model.init_model (Pylons 0.9.7) oder vorgenommen werden your_app.config.environment.load_environment (Pylons 1.0) Funktion - das sind das sind die Orte Ort, an dem die Engine-Instanz erstellt wird.
BEARBEITEN
Okay. Ich konnte die beschriebene Situation reproduzieren. Der obige Code benötigt einige Änderungen, um zu funktionieren. Unten ist, wie es gemacht werden sollte. Es spielt auch keine Rolle, ob es 0.9.7 oder 1.0 ist.
Sie müssen your_app/config/environment.py bearbeiten. Fügen Sie diese Exporte oben in die Datei ein:
import sqlalchemy
import sqlalchemy.interfaces
import _mysql_exceptions
Und das Ende der Funktion load_environment sollte so aussehen:
class MyListener(sqlalchemy.interfaces.PoolListener):
def __init__(self):
self.retried = False
def checkout(self, dbapi_con, con_record, con_proxy):
try:
dbapi_con.cursor().execute('select now()')
except _mysql_exceptions.OperationalError:
if self.retried:
self.retried = False
raise
self.retried = True
raise sqlalchemy.exc.DisconnectionError
config['sqlalchemy.listeners'] = [MyListener()]
engine = engine_from_config(config, 'sqlalchemy.')
init_model(engine)
Diesmal konnte ich es testen (auf Pylons 1.0 + SQLAlchemy 0.6.1) und es funktioniert. :)
