Manejo eficiente de sesiones usando el patrón de repositorio en FastAPI

Escribí un artículo hace unos cinco meses sobre la adaptación del patrón de repositorio con FastAPI, y obtuve muchas lecturas (gracias). He venido a escribir sobre una forma eficiente de manejar el manejo de sesiones, aún usando el patrón de repositorio.

Antes de pasar directamente a esto, noté que en producción, obtengo cualquiera de los siguientes errores cada vez que mis API intentan realizar una transacción que implica leer o escribir en la base de datos:

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

sqlalchemy.exc.PendingRollbackError

Este error indica que hay una transacción no confirmada en curso que debe revertirse antes de continuar con cualquier otra operación de la base de datos.

La causa más común de este error es una excepción no controlada que se produce durante una transacción de la base de datos, lo que impide que la transacción se confirme o revierta correctamente.

sqlalchemy.exc.InvalidRequestError

Este error indica que una operación o solicitud que realizó a la base de datos no es válida o no se admite. Puede haber varias causas para este error, que incluyen:

• Uso incorrecto de la API de SQLAlchemy, como llamar a un método o acceder a un atributo que no existe o no es aplicable en el contexto dado.

• Estructura de consulta o sintaxis SQL incorrecta.

• Asignación faltante o incorrecta entre los objetos de Python y las tablas/columnas de la base de datos.

Estoy seguro de que tiene ideas sobre cómo se pueden resolver estos errores, sin embargo, me gustaría decir que, aunque identifiqué cuál era el problema y lo solucioné, el problema persiste.

Si tiene curiosidad sobre cómo soluciono y resuelvo los problemas, puede considerar seguir los siguientes pasos:

sqlalchemy.exc.PendingRollbackError:

• Revise su código en busca de excepciones no controladas que puedan impedir que las transacciones se confirmen o reviertan correctamente. Asegúrese de manejar adecuadamente las excepciones y de confirmar o revertir la transacción según sea necesario.

• Compruebe si hay transacciones anidadas o de ejecución prolongada que puedan dar lugar a reversiones pendientes. Asegúrese de que todas las transacciones se comprometan o reviertan de manera oportuna.

• Revise su código para cualquier caso en el que pueda estar iniciando una nueva transacción antes de manejar correctamente la transacción anterior.

sqlalchemy.exc.InvalidRequestError:

• Revise el mensaje de error específico proporcionado con la excepción para identificar la causa de la solicitud no válida. Puede darle pistas sobre qué parte de su código o declaración SQL está causando el problema.

• Verifique su código SQLAlchemy en busca de llamadas a métodos incorrectos, accesos a atributos o uso incorrecto de la API.

• Revise sus instrucciones SQL para asegurarse de que tengan la sintaxis y la estructura correctas.

• Verifique que el esquema de su base de datos y las asignaciones de SQLAlchemy estén sincronizadas. Asegúrese de que existan todas las tablas y columnas requeridas y que sus objetos de Python estén correctamente asignados a las entidades de base de datos correspondientes.

Veamos cómo trabajé en una solución permanente que ha demostrado funcionar para mí. Procederé a usar un proyecto en el que trabajé cuando demostraba cómo usar patrones de repositorio.

Teníamos un módulo en el que almacenamos nuestra combinación de sesión base de orm con los siguientes códigos:

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

El problema con esta solución era que si ocurre una excepción en el proceso de una transacción (esto podría ser cualquier cosa: crear un usuario, financiar su billetera, etc.), las excepciones no se manejan correctamente y la sesión de la base de datos en tránsito no obtener reversión.

Después de tres meses de depuración y parches y mucha investigación, finalmente pude construir una forma eficiente de manejar sesiones.

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

En este código:

• DatabaseSession es una clase de administrador de contexto que maneja la sesión y garantiza que se cierre correctamente y se revierta en caso de error.

• El método __enter__ inicializa la sesión y la devuelve.

• El método __exit__ comprueba las excepciones y revierte la sesión si se produce una excepción. Luego cierra la sesión y la elimina de la sesión del ámbito.

• use_database_session es una función de utilidad que se puede usar como decorador o administrador de contexto para simplificar el uso de la sesión.

Aquí hay un ejemplo de cómo puede usar la función de utilidad 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.

El enfoque anterior proporciona una forma más limpia y eficiente de manejar las sesiones y garantiza que se reviertan o cierren correctamente en caso de error. Procedamos a cómo implementa el patrón de repositorio mientras usa la sesión de la base de datos en el 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

Lo siguiente sería integrar el repositorio anterior en la capa de servicio de su aplicación. Suponga que tiene una función de servicio que crea cuentas de usuarios; así es como lo haría usando nuestro nuevo método:

 # 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

El patrón anterior le permitirá encapsular las operaciones de la base de datos dentro de las clases del repositorio mientras aprovecha la sesión de la base de datos heredada. También proporciona una separación limpia entre sus modelos ORM y la lógica del repositorio.

Conclusión

En conclusión, el manejo eficiente de las sesiones es importante cuando se construyen sistemas back-end.

Los errores como sqlalchemy.exc.PendingRollbackError y sqlalchemy.exc.InvalidRequestError que ocurren durante las transacciones de la base de datos pueden generar inconsistencias en los datos y fallas en la aplicación si no se manejan correctamente.

Identificar y resolver estos errores es importante para mantener la integridad y confiabilidad del sistema.

Para abordar los problemas relacionados con el manejo de sesiones, es esencial implementar estrategias sólidas. Un enfoque es usar administradores de contexto, como DatabaseSessionMixin que demostramos en el artículo.

Este administrador de contexto garantiza que las sesiones se abran, cierren y reviertan correctamente en caso de excepciones. Al encapsular la lógica de la sesión dentro del administrador de contexto, puede optimizar la administración de la sesión y mejorar el manejo de errores.

Además, la integración del patrón de repositorio en la capa de servicio de la aplicación puede mejorar aún más la eficiencia del manejo de sesiones.

Al separar las operaciones de la base de datos en clases de repositorio y aprovechar la sesión heredada del administrador de contexto, puede lograr una organización de código más limpia y mantener una separación clara entre los modelos ORM y la lógica del repositorio.

En general, el manejo eficiente de sesiones es crucial para mantener la consistencia de los datos, prevenir errores y garantizar la estabilidad de los sistemas de back-end.

Al seguir las mejores prácticas, como el uso de administradores de contexto y la adopción del patrón de repositorio, los desarrolladores pueden crear sistemas robustos y confiables que administren sesiones y manejen errores de manera efectiva durante las transacciones de la base de datos.

Estoy abierto a escribir trabajos y buscar activamente roles de contrato que impliquen construir con Python (Django, FastAPI, etc.).