Gestion efficace des sessions à l'aide du modèle de référentiel dans FastAPI

J'ai écrit un article il y a environ cinq mois sur l'adaptation du modèle de référentiel avec FastAPI, et j'ai eu beaucoup de lectures (merci). Je suis venu écrire sur un moyen efficace de gérer la gestion des sessions, toujours en utilisant le modèle de référentiel.

Avant de me lancer directement, j'ai remarqué qu'en production, j'obtiens l'une des erreurs suivantes chaque fois que mes API tentent d'effectuer une transaction impliquant la lecture ou l'écriture dans la base de données :

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

sqlalchemy.exc.PendingRollbackError

Cette erreur indique qu'une transaction non validée en cours doit être annulée avant de poursuivre toute autre opération de base de données.

La cause la plus courante de cette erreur est une exception non gérée se produisant lors d'une transaction de base de données, qui empêche la transaction d'être validée ou annulée correctement.

sqlalchemy.exc.InvalidRequestError

Cette erreur indique qu'une opération ou une demande que vous avez effectuée auprès de la base de données n'est pas valide ou n'est pas prise en charge. Il peut y avoir diverses causes à cette erreur, notamment :

• Utilisation incorrecte de l'API SQLAlchemy, telle que l'appel d'une méthode ou l'accès à un attribut qui n'existe pas ou n'est pas applicable dans le contexte donné.

• Syntaxe SQL ou structure de requête incorrecte.

• Mappage manquant ou incorrect entre les objets Python et les tables/colonnes de la base de données.

Je suis certain que vous avez des idées sur la façon dont ces erreurs peuvent être résolues, cependant, je tiens à dire que même si j'ai identifié le problème et apporté une solution, le problème persiste.

Si vous êtes curieux de savoir comment je les dépanne et les résous, vous pouvez envisager de suivre les étapes suivantes :

sqlalchemy.exc.PendingRollbackError :

• Passez en revue votre code pour toutes les exceptions non gérées qui pourraient empêcher les transactions d'être validées ou annulées correctement. Assurez-vous que vous gérez correctement les exceptions et validez ou annulez la transaction si nécessaire.

• Recherchez les transactions de longue durée ou imbriquées susceptibles d'entraîner des restaurations en attente. Assurez-vous que toutes les transactions sont validées ou annulées en temps opportun.

• Passez en revue votre code pour tous les cas où vous pourriez démarrer une nouvelle transaction avant de traiter correctement la transaction précédente.

sqlalchemy.exc.InvalidRequestError :

• Passez en revue le message d'erreur spécifique fourni avec l'exception pour identifier la cause de la demande non valide. Cela peut vous donner des indices sur la partie de votre code ou de votre instruction SQL à l'origine du problème.

• Vérifiez votre code SQLAlchemy pour tout appel de méthode incorrect, accès aux attributs ou utilisation incorrecte de l'API.

• Vérifiez vos instructions SQL pour vous assurer qu'elles ont la syntaxe et la structure correctes.

• Vérifiez que votre schéma de base de données et les mappages SQLAlchemy sont synchronisés. Assurez-vous que toutes les tables et colonnes requises existent et que vos objets Python sont correctement mappés aux entités de base de données correspondantes.

Voyons comment j'ai travaillé sur une solution permanente qui s'est avérée efficace pour moi. Je vais continuer à utiliser un projet sur lequel j'ai travaillé lorsque je montrais comment utiliser les modèles de référentiel.

Nous avions un module où nous stockions notre mixin de session de base orm avec les codes suivants :

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

Le problème avec cette solution était que si une exception se produisait dans le processus d'une transaction (cela pouvait être n'importe quoi : créer un utilisateur, financer votre portefeuille, etc.) - les exceptions ne sont pas correctement gérées et la session de base de données en transit ne fonctionne pas. obtenir une restauration.

Après trois mois de débogage et de correctifs et de nombreuses recherches, j'ai finalement pu créer un moyen efficace de gérer les sessions.

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

Dans ce code :

• DatabaseSession est une classe de gestionnaire de contexte qui gère la session et s'assure qu'elle est correctement fermée et annulée en cas d'erreur.

• La méthode __enter__ initialise la session et la renvoie.

• La méthode __exit__ vérifie les exceptions et annule la session si une exception s'est produite. Il ferme ensuite la session et la supprime de la session délimitée.

• use_database_session est une fonction utilitaire qui peut être utilisée comme décorateur ou gestionnaire de contexte pour simplifier l'utilisation de la session.

Voici un exemple d'utilisation de la fonction utilitaire use_database_session :

 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.

L'approche ci-dessus fournit un moyen plus propre et plus efficace de gérer les sessions et garantit qu'elles sont correctement annulées ou fermées en cas d'erreur. Passons à la manière dont vous implémentez le modèle de référentiel lors de l'utilisation de la session de base de données dans l'ORM.

 # 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

Ensuite, il faudrait intégrer le référentiel ci-dessus dans la couche de service de votre application. Supposons que vous ayez une fonction de service qui crée un compte utilisateur ; voici comment vous le feriez en utilisant notre nouvelle méthode :

 # 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

Le modèle ci-dessus vous permettra d'encapsuler les opérations de base de données dans les classes de référentiel tout en tirant parti de la session de base de données héritée. Il fournit également une séparation nette entre vos modèles ORM et la logique du référentiel.

Conclusion

En conclusion, une gestion efficace des sessions est importante lors de la création de systèmes backend.

Les erreurs telles que sqlalchemy.exc.PendingRollbackError et sqlalchemy.exc.InvalidRequestError qui se produisent lors des transactions de base de données peuvent entraîner des incohérences de données et des échecs d'application si elles ne sont pas gérées correctement.

L'identification et la résolution de ces erreurs sont importantes pour maintenir l'intégrité et la fiabilité du système.

Pour résoudre les problèmes liés à la gestion des sessions, il est essentiel de mettre en œuvre des stratégies robustes. Une approche consiste à utiliser des gestionnaires de contexte, tels que le DatabaseSessionMixin que nous avons démontré dans l'article.

Ce gestionnaire de contexte garantit que les sessions sont correctement ouvertes, fermées et annulées en cas d'exceptions. En encapsulant la logique de session dans le gestionnaire de contexte, vous pouvez rationaliser la gestion des sessions et améliorer la gestion des erreurs.

De plus, l'intégration du modèle de référentiel dans la couche de service de l'application peut encore améliorer l'efficacité de la gestion des sessions.

En séparant les opérations de base de données en classes de référentiel et en tirant parti de la session héritée du gestionnaire de contexte, vous pouvez obtenir une organisation de code plus propre et maintenir une séparation claire entre les modèles ORM et la logique du référentiel.

Dans l'ensemble, une gestion efficace des sessions est essentielle pour maintenir la cohérence des données, prévenir les erreurs et garantir la stabilité des systèmes backend.

En suivant les meilleures pratiques, telles que l'utilisation de gestionnaires de contexte et l'adoption du modèle de référentiel, les développeurs peuvent créer des systèmes robustes et fiables qui gèrent efficacement les sessions et gèrent les erreurs lors des transactions de base de données.

Je suis ouvert à l'écriture de concerts et à la recherche active de rôles contractuels impliquant la construction avec Python (Django, FastAPI, etc.).