Fixing SQLAlchemy PendingRollbackError: Stop the 'Poisoned Session' Cycle

Why Your Database Is 'Ghosting' You

Few things are more frustrating than a database driver that refuses to run a query because of a mistake made three steps ago. You run a query, it fails, and suddenly every subsequent call hits a brick wall. This is the PendingRollbackError.

sqlalchemy.exc.PendingRollbackError: Can't reconnect until invalid transaction is rolled back

The Root Cause: Your Session Is Poisoned

At its core, this error is a safety shut-off valve. It happens when a previous operation—perhaps a UniqueConstraint violation on a user's email or a 5-second query timeout—fails, but the session stays open.

SQLAlchemy marks the current transaction as "inactive" or "failed" the moment an exception occurs. Because the database connection is still tied to that failed state, SQLAlchemy blocks new commands to prevent you from accidentally committing partial or corrupt data. It won't let you proceed until you acknowledge the failure.

You will likely encounter this in three specific situations:

• The Forgotten Rollback: You caught an exception in a try/except block but didn't tell the session to reset.
• Thread Contamination: You're using a global session where one thread's crash breaks the connection for everyone else.
• Zombie Tasks: Background workers like Celery are reusing a broken session across different jobs.

Fix 1: The Manual Safety Net

The most reliable fix is to ensure every session.commit() lives inside a try/except block. If the commit fails, the rollback() clears the "poisoned" state immediately.

from sqlalchemy.exc import SQLAlchemyError

def create_user_profile(session, profile_data):
    try:
        session.add(profile_data)
        session.commit()
    except SQLAlchemyError as e:
        # This clears the failed state so the next request works
        session.rollback()
        print(f"Database error: {e}")
        raise
    finally:
        session.close()

Fix 2: Modern Context Managers (SQLAlchemy 2.0 Style)

If you are using SQLAlchemy 1.4 or 2.0, stop managing commits manually. Use the begin() context manager. It is cleaner, safer, and handles the rollback for you if anything goes wrong.

from sqlalchemy.orm import Session

# The session automatically rolls back if an error occurs inside the 'with' block
with Session(engine) as session:
    with session.begin():
        # Let's say this triggers a DuplicateKey error
        session.add(User(email="duplicate@example.com")) 

# Once the block exits, the session is clean and ready for the next task.

Fix 3: Bulletproofing FastAPI Dependencies

In FastAPI, this error usually crops up because a failed request leaves a connection "dirty" in the connection pool. The next user who gets that connection inherits the error. Use a yield pattern to ensure every request starts with a fresh slate.

from fastapi import Depends
from database import SessionLocal

def get_db_session():
    db = SessionLocal()
    try:
        yield db
    except Exception:
        # If the route crashes, we roll back here
        db.rollback()
        raise
    finally:
        db.close()

@app.post("/register")
def register_user(user: UserSchema, db: Session = Depends(get_db_session)):
    db.add(user)
    db.commit()
    return {"message": "User created"}

Fix 4: Thread Safety with Scoped Sessions

Are you building a multi-threaded app? Use scoped_session. This gives each thread its own isolated session. If Thread A crashes, Thread B keeps running without ever seeing a PendingRollbackError.

from sqlalchemy.orm import scoped_session, sessionmaker

session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

# Thread-local session access
local_session = Session()
try:
    # Perform DB work
    local_session.commit()
except:
    local_session.rollback()
finally:
    # Always remove the session to return the connection to the pool
    Session.remove() 

How to Verify the Fix

Don't guess if it's fixed. Prove it. Run a small script that intentionally breaks a transaction, then tries to run a valid one immediately after.

• Try to insert a record with a duplicate Primary Key (e.g., ID 101).
• Catch the IntegrityError.
• Run session.execute("SELECT 1").

If SELECT 1 returns a result, your rollback logic is solid. If it throws PendingRollbackError, your session is still poisoned.

# Quick Test Script
try:
    session.add(User(id=1)) # Trigger a crash
    session.commit()
except Exception:
    session.rollback() # The vital fix

# If this works, you've solved it
check = session.execute("SELECT 1").scalar()
print(f"Session Status: {'Healthy' if check == 1 else 'Broken'}")