Efficient Session Handling Using the Repository Pattern in FastAPI

I wrote an article about five months ago about adapting the repository pattern with FastAPI, and I got a whole lot of reads (thank you). I have come to write about an efficient way to handle session handling, still using the repository pattern.

Before I jump right into it, I noticed that on production, I get either of the following errors whenever my APIs are trying to make a transaction that involves reading or writing to the database:

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

sqlalchemy.exc.PendingRollbackError

This error indicates that there is an uncommitted transaction in progress that needs to be rolled back before proceeding with any other database operations.

The most common cause of this error is an unhandled exception occurring during a database transaction, which prevents the transaction from being committed or rolled back properly.

sqlalchemy.exc.InvalidRequestError

This error indicates that an operation or request that you made to the database is invalid or not supported. There can be various causes for this error, including:

• Incorrect usage of SQLAlchemy API, such as calling a method or accessing an attribute that doesn't exist or is not applicable in the given context.

• Incorrect SQL syntax or query structure.

• Missing or incorrect mapping between Python objects and database tables/columns.

I am certain that you have thoughts about how these errors can be resolved, however, I would like to state that even though I identified what the problem was and made a fix- the problem persists.

If you are curious as to how I troubleshoot and resolve them, you can consider following the following steps:

sqlalchemy.exc.PendingRollbackError:

• Review your code for any unhandled exceptions that might prevent transactions from being committed or rolled back properly. Ensure that you are properly handling exceptions and either committing or rolling back the transaction as needed.

• Check for any long-running or nested transactions that might lead to pending rollbacks. Make sure that all transactions are committed or rolled back in a timely manner.

• Review your code for any cases where you might be starting a new transaction before properly handling the previous transaction.

sqlalchemy.exc.InvalidRequestError:

• Review the specific error message provided with the exception to identify the cause of the invalid request. It might give you hints about what part of your code or SQL statement is causing the issue.

• Check your SQLAlchemy code for any incorrect method calls, attribute accesses, or incorrect usage of the API.

• Review your SQL statements to ensure they have the correct syntax and structure.

• Verify that your database schema and SQLAlchemy mappings are in sync. Ensure that all required tables and columns exist and that your Python objects are properly mapped to the corresponding database entities.

Let’s get right into how I worked on a permanent solution that has proven to work for me. I shall proceed to use a project that I worked on when I was demonstrating how to use repository patterns.

We had a module where we stored our orm base session mixin with the following codes:

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

The problem with this solution was that if an exception occurs in the process of a transaction (this could be anything: creating a user, funding your wallet, etc.)- the exceptions are not properly handled, and the database session in transit does not get rollback.

After three months of debugging and patching and lots of research, I finally was able to build an efficient way of handling 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()

In this code:

• DatabaseSession is a context manager class that handles the session and ensures that it is properly closed and rolled back in case of an error.

• __enter__ method initializes the session and returns it.

• __exit__ method checks for exceptions and rolls back the session if an exception occurred. It then closes the session and removes it from the scoped session.

• use_database_session is a utility function that can be used as a decorator or context manager to simplify session usage.

Here's an example of how you can use the use_database_session utility function:

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.

The above approach provides a cleaner and more efficient way to handle sessions and ensures that they are properly rolled back or closed in case of an error. Let’s proceed to how you implement the repository pattern while using the database session in the 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

Next would be to integrate the above repository into the service layer of your application. Suppose you have a service function that creates users account; here’s how you’d do it using our new method:

# 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

The above pattern will allow you to encapsulate database operations within repository classes while leveraging the inherited database session. It also provides a clean separation between your ORM models and the repository logic.

Conclusion

In conclusion, efficient handling of sessions is important when building backend systems.

The errors like sqlalchemy.exc.PendingRollbackError and sqlalchemy.exc.InvalidRequestError that occur during database transactions can lead to data inconsistencies and application failures if not handled properly.

Identifying and resolving these errors is important for maintaining the integrity and reliability of the system.

To address the issues related to session handling, it is essential to implement robust strategies. One approach is to use context managers, such as the DatabaseSessionMixin we demonstrated in the article.

This context manager ensures that sessions are properly opened, closed, and rolled back in case of exceptions. By encapsulating the session logic within the context manager, you can streamline session management and improve error handling.

Additionally, integrating the repository pattern into the service layer of the application can further enhance the efficiency of session handling.

By separating the database operations into repository classes and leveraging the inherited session from the context manager, you can achieve cleaner code organization and maintain a clear separation between ORM models and repository logic.

Overall, efficient session handling is crucial for maintaining data consistency, preventing errors, and ensuring the stability of backend systems.

By following best practices, such as using context managers and adopting the repository pattern, developers can build robust and reliable systems that effectively manage sessions and handle errors during database transactions.

I am open to writing gigs and actively searching for contract roles that involve building with Python (Django, FastAPI, etc.).