Warum ist die SQLalchemy -Datenbankverbindung nicht korrekt geschlossen? Wie löst ich dieses Problem?

Richtige Methode zum Schließen der SQLalchemy -Datenbankverbindung und zur Fehlerbehebung

Bei der Verwendung von Pythons SQLalchemy -Bibliothek für Datenbankvorgänge ist es wichtig sicherzustellen, dass die Datenbankverbindung ordnungsgemäß geschlossen ist, um Ressourcenlecks und Leistungsprobleme zu vermeiden. Dieser Artikel analysiert ein gemeinsames Problem der SQLAlchemy -Verbindung und liefert Lösungen.

Das folgende Code -Snippet zeigt ein Beispiel für mögliche Probleme mit Verbindungsschließungen:

 Von SQLAlchemy Import create_engine, URL, löschen, aktualisieren, auswählen, existiert
von SQLALCHEMY.ORM Import Sessionmaker, Scoped_Session
von core.database.base importieren Basi
aus Flask Import g, current_app

Importlib import
Import Re


Klassendatenbank: # Ändern Sie den Klassennamen, um den Anfangsbuchstaben zu profitieren, und entsprechen der Python -Spezifikation env = Keine

    Def Set (Selbst, Schlüssel: STR, Wert: Any):
        "" "
        Legen Sie den Eigenschaftswert auf G.Application oder G.Platform gemäß der Umgebungsvariablen fest
        "" "
        Wenn self.env == "Anwendung":
            g.application = self.container._replace (** {key: value})
        elif self.env == 'Plattform':
            g.Platform = self.container._replace (** {Schlüssel: Wert})

    @Eigentum
    Def Container (Selbst):
        "" "
        Gibt G.Application oder G.Platform Container "" "" "" "" "
        Wenn self.env == "Anwendung":
            Wenn "Anwendung" nicht in G:
                g.application = type.application (keine, keine, keine)
            Return G. Application
        elif self.env == 'Plattform':
            Wenn "Plattform" nicht in G:
                g.Platform = type.platform (keine, keine)
            RECHT G.PLATFORM

    @Eigentum
    Def Database_conf (Selbst):
        "" "
        Datenbankkonfiguration "" "" ""
        return base.Setting (current_app.config ["Datenbank"])

    @Eigentum
    def __database_core (self):
        "" "
        Erstellen Sie eine Datenbanksitzung und richten Sie sie auf das Instanzattribut "" "" "" "" "."
        Wenn nicht Hasattr (Selbst, '_database_core'):
            self._database_core = self .__ Create_Session (** self.database_conf)
        return self._database_core

    @Eigentum
    def __create_engine (self):
        "" "
        Holen Sie sich die Datenbank -Engine und unterbinden Sie sie in das Instanzattribut "" ""
        Return Self .__ Database_core.Engine

    @Eigentum
    def __create_database (self):
        "" "
        Holen Sie sich die Datenbanksitzung und unterbinden Sie sie in das Instanzattribut "" ""
        Return self .__ Database_core.Session

    def __create_session (self, ** config):
        "" "
        Erstellen Sie eine Datenbanksitzung "" ""
        Engine = self.create_engine (** config)
        Session = Scoped_Session (SessionMaker (Bind = Engine, autoflush = true))
        Rückgabetyp.Database (Engine = Engine, Session = Session ())

    @ClassMethod
    Def create_engine (cls, ** kwargs):
        "" "
        Erstellen Sie eine Datenbank -Engine "" ""
        return create_engine (url.create ("mysql pymysql", ** kwargs), echo = true, isolation_level = "autocommit"))

    @StaticMethod
    Def create_all (Modelle: Liste, Engine = Keine):
        "" "
        Erstellen Sie Tabellen für alle Modelle "" "
        Tabellen = [database.get_model (Modell) .__ Tabelle__ für Modell in Modellen]
        Base.metadata.create_all (Bind = Motor, Tabellen = Tabellen)

    Def create_table (Selbst, Tabellen: Liste):
        "" "
        Erstellen Sie eine Tabelle für das angegebene Modell "" "
        Datenbank.Create_all (Models = Tabellen, Engine = self .__ Create_Engine)

    @StaticMethod
    Def get_model (Modell: STR):
        "" "
        Holen Sie sich das Modellobjekt "" "
        module = importlib.import_module (f "modell. {model.split ('_') [0]}. {Modell}")
        class_name = '' .join (re.findall (r "[a-za-z]", model.split (".") [-1] .title ())))
        Getattr zurückgeben (Modul, class_name) ())

    @Eigentum
    Def -Datenbank (Selbst):
        "" "
        Datenbanksitzung erhalten "" ""
        Return self .__ create_database

    Def table_data_query_all (Selbst, Modell: Any, Bedingung: List = Keine, Reihenfolge: list = none, limit: int = 500,
                             Felder: list = None) -> Liste [Diktat]:
        "" "
        Abfragen Sie alle Daten "" ""
        query = select (Modell)
        Wenn Felder:
            query = query.with_only_columns (*fields)
        Wenn Zustand:
            query = query.filter (*Bedingung)
        Wenn bestellen:
            query = query.order_by (*order)
        Ergebnisse = [row.dict () für Zeile in self.database.execute (query.limit (limit)). scalars ()]]
        Rückgabeergebnisse

    DEF table_data_query_one (Selbst, Modell: Any, Bedingung: list = Keine) -> Diktat:
        "" "
        Fragen Sie ein einzelnes Datenstück "" "" "
        result = self.database.execute (select (modell) .filter (*Bedingung) .Limit (1)). scalar_one_or_none ()
        Rückgabe keine Wenn das Ergebnis kein anderes Ergebnis ist.dict ()

    DEF table_data_query_exists (Selbst, Bedingung: Liste) -> bool:
        "" "
        Fragen Sie, ob die Daten "" "" ""
        return self.database.Query (existiert (). Wobei (*Zustand)) scalar ()

    DEF table_data_insert_all (Selbst, Modelle: Liste) -> Keine:
        "" "
        Batch -Einfügung von Daten "" ""
        mit self.database als db:
            db.add_all (Modelle)
            db.commit ()

    Def table_data_insert_one (self, modell, Daten: bool = false) -> int | DICT:
        "" "
        Fügen Sie ein einzelnes Datenstück "" "ein
        mit self.database als db:
            db.add (Modell)
            db.commit ()
            return model.dict () wenn data sonst model.id

    DEF TABELLE_DATA_UPDATE (Selbst, Modell: Any, Bedingung: Liste, Daten: Diktat) -> Keine:
        "" "
        Aktualisieren Sie Daten "" "
        mit self.database als db:
            db.execute (update (Modell) .Where (*Zustand) .Values ​​(** Daten)))
            db.commit () # def table_data_delete (self, model: angemessen, Bedingung: Liste) -> Keine:
        "" "
        Daten löschen "" ""
        mit self.database als db:
            db.execute (delete (Modell). Wo (*Zustand))
            db.commit () # def close (self):
        "" "
        Schließen Sie die Datenbankverbindung "" "
        Wenn Hasattr (Selbst, '_database_core'):
            self._database_core.session.close ())
            self._database_core.engine.dispose ())
            Del self._database_core

    def __del __ (selbst):
        "" "
        Destruktor stellen Sie sicher, dass die Verbindung geschlossen ist "" ""
        self.close ()

Verbesserungsanweisungen:

1. Klassennamenspezifikation: database in Database ändern und die Spezifikationen von Python -Namenszahlen einhalten.
2. Eigenschaftscache: Verwenden Sie @property und Instanzeigenschafts -Cache _database_core um eine wiederholte Sitzungserstellung zu vermeiden.
3. Explizites Commit: db.commit() in table_data_update und table_data_delete hinzugefügt, um Transaktions -Commits sicherzustellen.
4. Ressourcenveröffentlichung: close() ruft explizit session.close() und engine.dispose() um die Ressource freizugeben. del self._database_core löscht zwischengespeichertes Sitzungsobjekt.
5. Ausnahmeberechnung: Erwägen Sie try...except Block, um mögliche Ausnahmen zu verarbeiten, z. B. Datenbankverbindungsfehler.
6. Verwendung von scoped_session : scoped_session wird normalerweise in Flask -Anwendungen mit g -Objekten verwendet, um sicherzustellen, dass jede Anforderung eine unabhängige Sitzung verwendet und beim Ende der Anforderung automatisch geschlossen wird. Der Code spiegelt jedoch nicht das Kontextmanagement der Flaskanfragen wider, daher ist dispose() erforderlich. dispose() ist möglicherweise nicht erforderlich, wenn die Kontextverwaltung von Flask verwendet wird, aber session.close() ist noch erforderlich.

Lösung:

Das Hauptproblem ist die Verwendung von scoped_session und dem Zeitpunkt der Ressourcenfreigabe. scoped_session selbst garantiert keine automatische Schließung der Verbindung, sondern verwaltet nur den Umfang der Sitzung. self.database.get_bind().dispose() funktioniert in einigen Fällen möglicherweise nicht, da es die zugrunde liegende Datenbankverbindung möglicherweise nicht ordnungsgemäß schließt.

Daher ist es notwendig, gegebenenfalls die Methode close() aufzurufen oder close() im Destructor __del__ der Klasse aufzurufen, um sicherzustellen, dass die Verbindung korrekt geschlossen wird. Es ist jedoch keine bewährte Verfahren, sich auf __del__ zu verlassen, da Pythons Müllsammlungsmechanismus unvorhersehbar ist. Es wird empfohlen, instance.close() nach Verwendung Database .

Best Practices:

• Verwenden Sie einen Kontextmanager ( with Anweisung), um Datenbanksitzungen zu verwalten: Dies stellt sicher, dass die Sitzung nach Ausführung des Codeblocks automatisch geschlossen wird.
• In Flask-Anwendungen ist es möglich, Datenbankverbindungen und -sitzungen leichter zu verwalten. Diese Bibliotheken verarbeiten normalerweise das Schließen und Freigeben von Verbindungen automatisch.

Durch die obigen Verbesserungen kann das Problem, dass die SQLalchemy -Datenbankverbindung nicht korrekt geschlossen werden kann, effektiv gelöst und der Code robust und wartbar. Denken Sie daran, explizit Schließen von Verbindungen bester Verfahren und vermeiden Sie es, sich auf Müllsammlungsmechanismen zu verlassen.

Das obige ist der detaillierte Inhalt vonWarum ist die SQLalchemy -Datenbankverbindung nicht korrekt geschlossen? Wie löst ich dieses Problem?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!