Effiziente Sitzungsverwaltung mithilfe des Repository-Musters in FastAPI

Ich habe vor etwa fünf Monaten einen Artikel über die Anpassung des Repository-Musters mit FastAPI geschrieben und eine ganze Menge gelesen (danke). Ich bin gekommen, um über eine effiziente Möglichkeit zu schreiben, mit der Sitzungsverwaltung umzugehen und dabei weiterhin das Repository-Muster zu verwenden.

Bevor ich direkt loslege, ist mir aufgefallen, dass ich in der Produktion immer dann eine der folgenden Fehlermeldungen erhalte, wenn meine APIs versuchen, eine Transaktion durchzuführen, die das Lesen oder Schreiben in die Datenbank beinhaltet:

• sqlalchemy.exc.PendingRollbackError
• sqlalchemy.exc.InvalidRequestError

sqlalchemy.exc.PendingRollbackError

Dieser Fehler weist darauf hin, dass eine nicht festgeschriebene Transaktion ausgeführt wird, die zurückgesetzt werden muss, bevor mit anderen Datenbankvorgängen fortgefahren werden kann.

Die häufigste Ursache für diesen Fehler ist eine unbehandelte Ausnahme, die während einer Datenbanktransaktion auftritt und verhindert, dass die Transaktion ordnungsgemäß festgeschrieben oder zurückgesetzt wird.

sqlalchemy.exc.InvalidRequestError

Dieser Fehler weist darauf hin, dass ein Vorgang oder eine Anforderung, die Sie an die Datenbank gestellt haben, ungültig ist oder nicht unterstützt wird. Dieser Fehler kann verschiedene Ursachen haben, darunter:

• Falsche Verwendung der SQLAlchemy-API, z. B. Aufrufen einer Methode oder Zugriff auf ein Attribut, das nicht existiert oder im gegebenen Kontext nicht anwendbar ist.

• Falsche SQL-Syntax oder Abfragestruktur.

• Fehlende oder falsche Zuordnung zwischen Python-Objekten und Datenbanktabellen/-spalten.

Ich bin mir sicher, dass Sie darüber nachdenken, wie diese Fehler behoben werden können. Ich möchte jedoch darauf hinweisen, dass das Problem weiterhin besteht, obwohl ich das Problem erkannt und behoben habe.

Wenn Sie wissen möchten, wie ich sie behebe und behebe, können Sie die folgenden Schritte in Betracht ziehen:

sqlalchemy.exc.PendingRollbackError:

• Überprüfen Sie Ihren Code auf nicht behandelte Ausnahmen, die möglicherweise verhindern, dass Transaktionen ordnungsgemäß festgeschrieben oder zurückgesetzt werden. Stellen Sie sicher, dass Sie Ausnahmen ordnungsgemäß behandeln und die Transaktion je nach Bedarf festschreiben oder zurücksetzen.

• Suchen Sie nach lang laufenden oder verschachtelten Transaktionen, die zu ausstehenden Rollbacks führen könnten. Stellen Sie sicher, dass alle Transaktionen rechtzeitig festgeschrieben oder zurückgesetzt werden.

• Überprüfen Sie Ihren Code auf Fälle, in denen Sie möglicherweise eine neue Transaktion starten, bevor Sie die vorherige Transaktion ordnungsgemäß verarbeiten.

sqlalchemy.exc.InvalidRequestError:

• Überprüfen Sie die spezifische Fehlermeldung, die mit der Ausnahme bereitgestellt wird, um die Ursache der ungültigen Anforderung zu ermitteln. Es könnte Ihnen Hinweise darauf geben, welcher Teil Ihres Codes oder Ihrer SQL-Anweisung das Problem verursacht.

• Überprüfen Sie Ihren SQLAlchemy-Code auf falsche Methodenaufrufe, Attributzugriffe oder falsche Verwendung der API.

• Überprüfen Sie Ihre SQL-Anweisungen, um sicherzustellen, dass sie die richtige Syntax und Struktur haben.

• Stellen Sie sicher, dass Ihr Datenbankschema und Ihre SQLAlchemy-Zuordnungen synchron sind. Stellen Sie sicher, dass alle erforderlichen Tabellen und Spalten vorhanden sind und dass Ihre Python-Objekte ordnungsgemäß den entsprechenden Datenbankentitäten zugeordnet sind.

Kommen wir gleich dazu, wie ich an einer dauerhaften Lösung gearbeitet habe, die sich für mich bewährt hat. Ich werde weiterhin ein Projekt verwenden, an dem ich gearbeitet habe, als ich demonstrierte, wie man Repository-Muster verwendet.

Wir hatten ein Modul, in dem wir unser Orm-Basis-Session-Mixin mit den folgenden Codes gespeichert haben:

 # SQLAlchemy Imports from sqlalchemy.orm import Session # Own Imports from config.database import SessionLocal from core.settings import ledger_settings class ORMSessionMixin: """Base orm session mixin for interacting with the database.""" def __init__(self): """ Get the next database session from the database pool. """ self.orm: Session = self.get_db().__next__() def get_db(self): """ This method creates a database session, yields it, rollback the transaction if there's an exception and then finally closes the session. Yields: db: scoped database session """ db = SessionLocal() try: yield db except Exception: db.rollback() finally: db.close()

Das Problem bei dieser Lösung bestand darin, dass, wenn im Prozess einer Transaktion eine Ausnahme auftritt (dies kann alles sein: Erstellen eines Benutzers, Einzahlen in Ihr Wallet usw.), die Ausnahmen nicht ordnungsgemäß behandelt werden und die Datenbanksitzung während der Übertragung nicht ordnungsgemäß behandelt wird Rollback bekommen.

Nach drei Monaten des Debuggens und Patchens und viel Recherche war ich endlich in der Lage, eine effiziente Methode zur Sitzungsabwicklung zu entwickeln.

 # SQLAlchemy Imports import sqlalchemy from sqlalchemy.orm import Session # Own Imports from config.database.connection import SessionLocal class DatabaseSessionMixin: """Database session mixin.""" def __enter__(self) -> Session: self.db = SessionLocal() return self.db def __exit__(self, exc_type, exc_val, exc_tb): try: if exc_type is not None: self.db.rollback() except sqlalchemy.exc.SQLAlchemyError: pass finally: self.db.close() SessionLocal.remove() def use_database_session(): return DatabaseSessionMixin()

In diesem Code:

• DatabaseSession ist eine Kontextmanagerklasse, die die Sitzung verwaltet und sicherstellt, dass sie im Fehlerfall ordnungsgemäß geschlossen und zurückgesetzt wird.

• Die Methode __enter__ initialisiert die Sitzung und gibt sie zurück.

• Die Methode __exit__ prüft auf Ausnahmen und setzt die Sitzung zurück, wenn eine Ausnahme aufgetreten ist. Anschließend wird die Sitzung geschlossen und aus der bereichsbezogenen Sitzung entfernt.

• use_database_session ist eine Hilfsfunktion, die als Dekorator oder Kontextmanager verwendet werden kann, um die Sitzungsnutzung zu vereinfachen.

Hier ist ein Beispiel dafür, wie Sie die Dienstprogrammfunktion use_database_session verwenden können:

 with use_database_session() as db: # perform logic that uses the session # ... # After exiting the context, the session will be automatically closed and removed from the scoped session.

Der obige Ansatz bietet eine sauberere und effizientere Möglichkeit, Sitzungen zu verwalten und stellt sicher, dass sie im Fehlerfall ordnungsgemäß zurückgesetzt oder geschlossen werden. Fahren wir damit fort, wie Sie das Repository-Muster implementieren, während Sie die Datenbanksitzung im ORM verwenden.

 # SQLAlchemy Imports import sqlalchemy from sqlalchemy.orm import Session class BaseRepository: def __init__(self, session: Session): self.db = session class UserRepository(BaseRepository): """Operations to interact with the `users` table in the database.""" def get(self, user_id: int) -> User: """This method gets a user from the database.""" user = ( self.db.query(User) .filter(User.id == user_id) .first() ) return user def create(self, name: str, email: str, password: str) -> User: """This method creates a user.""" user = User(name=name, email=email, password=password) self.db.add(user) self.db.commit() self.db.refresh(user) return user def update_user(self, user_id: int, updated_data: dict): """This method updates a user.""" user = self.get(user_id) if user: for key, value in updated_data.items(): setattr(user, key, value) self.db.commit() return user return None def delete_user(self, user_id): """This method deletes a user.""" user = self.get_user(user_id) if user: self.db.delete(user) self.db.commit() return True return False

Als nächstes müssten Sie das obige Repository in die Serviceschicht Ihrer Anwendung integrieren. Angenommen, Sie verfügen über eine Dienstfunktion, die ein Benutzerkonto erstellt. So würden Sie es mit unserer neuen Methode machen:

 # Apps Imports from apps.users.models import User from apps.users.repo import UserRepository from apps.users.schemas.auth import UserCreate # Config Imports from config.security.hashers import password from config.database.session_mixin import use_database_session async def create_user(user: UserCreate) -> User: """ This function creates a new user in the database. :param user: schemas.UserCreate :type user: schemas.UserCreate :return: The user object """ with use_database_session() as db: users_repo = UserRepository(db) user = users_repo.create( user.name, user.email, password.hash(user.password) ) return user

Das obige Muster ermöglicht es Ihnen, Datenbankoperationen in Repository-Klassen zu kapseln und gleichzeitig die geerbte Datenbanksitzung zu nutzen. Es sorgt außerdem für eine saubere Trennung zwischen Ihren ORM-Modellen und der Repository-Logik.

Abschluss

Zusammenfassend lässt sich sagen, dass beim Aufbau von Backend-Systemen eine effiziente Sitzungsabwicklung wichtig ist.

Fehler wie sqlalchemy.exc.PendingRollbackError “ und sqlalchemy.exc.InvalidRequestError “, die bei Datenbanktransaktionen auftreten, können bei unsachgemäßer Behandlung zu Dateninkonsistenzen und Anwendungsfehlern führen.

Das Erkennen und Beheben dieser Fehler ist wichtig für die Aufrechterhaltung der Integrität und Zuverlässigkeit des Systems.

Um die Probleme im Zusammenhang mit der Sitzungsbearbeitung anzugehen, ist die Implementierung robuster Strategien unerlässlich. Ein Ansatz besteht darin, Kontextmanager zu verwenden, wie z. B. den DatabaseSessionMixin den wir im Artikel gezeigt haben.

Dieser Kontextmanager stellt sicher, dass Sitzungen im Falle von Ausnahmen ordnungsgemäß geöffnet, geschlossen und zurückgesetzt werden. Durch die Kapselung der Sitzungslogik im Kontextmanager können Sie die Sitzungsverwaltung optimieren und die Fehlerbehandlung verbessern.

Darüber hinaus kann die Integration des Repository-Musters in die Serviceschicht der Anwendung die Effizienz der Sitzungsverwaltung weiter verbessern.

Durch die Aufteilung der Datenbankoperationen in Repository-Klassen und die Nutzung der vom Kontextmanager geerbten Sitzung können Sie eine sauberere Codeorganisation erreichen und eine klare Trennung zwischen ORM-Modellen und Repository-Logik aufrechterhalten.

Insgesamt ist eine effiziente Sitzungsabwicklung entscheidend für die Aufrechterhaltung der Datenkonsistenz, die Vermeidung von Fehlern und die Gewährleistung der Stabilität von Backend-Systemen.

Durch die Befolgung von Best Practices wie der Verwendung von Kontextmanagern und der Übernahme des Repository-Musters können Entwickler robuste und zuverlässige Systeme erstellen, die Sitzungen effektiv verwalten und Fehler bei Datenbanktransaktionen behandeln.

Ich bin offen für das Schreiben von Aufträgen und suche aktiv nach Vertragsrollen, die das Erstellen mit Python (Django, FastAPI usw.) beinhalten.