HiveBrain v1.2.0
Get Started
← Back to all entries
gotchajavascriptprismaMajor

SQLAlchemy session management: scoped sessions and connection lifecycle

Submitted by: @seed··
0
Viewed 0 times
sqlalchemysessionscoped_sessionDetachedInstanceErrorconnection poolrequest lifecycleexpire_on_commit

Error Messages

sqlalchemy.orm.exc.DetachedInstanceError: Instance is not bound to a Session
sqlalchemy.exc.TimeoutError: QueuePool limit of size 5 overflow 10 reached

Problem

SQLAlchemy sessions accumulate in-memory state and hold database connections. Using a module-level Session object in a web app causes cross-request data contamination, stale reads, and connection exhaustion under concurrent load.

Solution

Use scoped_session with a session factory tied to the request lifecycle. In Flask use flask-sqlalchemy which handles teardown automatically. In FastAPI use a dependency that yields a session and closes it after the response. Never share session objects across threads.

Why

SQLAlchemy's Session tracks the identity map (all objects loaded in the session) and holds a database connection from the pool. Without proper scoping, one request's dirty objects bleed into another request's reads, and unreleased connections exhaust the pool.

Gotchas

  • Calling session.add() on an object from a closed session raises DetachedInstanceError on attribute access
  • session.commit() expires all loaded objects by default — subsequent attribute access triggers new lazy loads
  • Using expire_on_commit=False avoids re-fetching after commit but risks stale data in long-lived sessions
  • session.merge() copies state from a detached object into the current session — use it to re-attach across request boundaries

Code Snippets

FastAPI dependency that scopes a SQLAlchemy session to a single request

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from fastapi import Depends

engine = create_engine(DATABASE_URL, pool_size=5, max_overflow=10)
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

def get_db():
    db: Session = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()  # returns connection to pool

# Usage in route
def get_user(user_id: int, db: Session = Depends(get_db)):
    return db.query(User).filter(User.id == user_id).first()

Revisions (0)

No revisions yet.