SQLAlchemy ORM vs Core — When to Use Which
Think of an SQLAlchemy ORM session as a private, in-memory cache and transaction manager for your Python objects. Changes made outside this session (like Core updates) aren't automatically reflected in its cached objects until explicitly told to refresh.
The Setup
You are migrating an ingestion system that loads millions of metrics daily. To optimize performance, you bypass the ORM and execute a raw core bulk update, but subsequent queries in the same transaction still read stale data.
What Does This Print?
from sqlalchemy import create_engine, select, update, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# Insert initial record
user = User(id=1, name="Alice")
session.add(user)
session.commit()
# Bypass ORM: Update record directly via Core execution
stmt = update(User.__table__).where(User.id == 1).values(name="Bob")
session.execute(stmt)
# Query the record again using the ORM session
user_refetched = session.get(User, 1)
print(f"ORM User Name: {user_refetched.name}")The Output
Even though the Core update statement was executed successfully on the database, the ORM identity map still holds the original state of the User object (Alice). SQLAlchemy's identity map acts as an in-memory cache for objects loaded within the active transaction. When you execute a direct Core update statement, the ORM state tracker is completely bypassed. Consequently, calling session.get() skips the database roundtrip entirely, retrieving the stale object directly from memory.
Why Python Does This
SQLAlchemy uses the Unit of Work pattern where the session maintains an Identity Map (a Python dictionary mapping (Mapper, (primary_key,)) to object instances). This guarantees that a single row has exactly one unique object instance in memory, preserving object identity (user_refetched is user). When executing a Core statement via session.execute(update(...)), SQLAlchemy compiles and executes raw SQL directly. Because this does not involve ORM lifecycle events, the session does not invalidate or update its internal dictionaries. To fix this, you must explicitly instruct SQLAlchemy to synchronize the session state during direct statement execution.
The Fix
from sqlalchemy import create_engine, update, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
user = User(id=1, name="Alice")
session.add(user)
session.commit()
# FIX: Synchronize session state during direct Core execution
stmt = update(User.__table__).where(User.id == 1).values(name="Bob")
session.execute(stmt, execution_options={"synchronize_session": "fetch"})
user_refetched = session.get(User, 1)
print(f"ORM User Name: {user_refetched.name}") # Output: BobPassing execution_options={"synchronize_session": "fetch"} instructs SQLAlchemy to perform a SELECT before executing the UPDATE, fetching the primary keys of all rows that match the WHERE clause. It then walks the session's identity map and updates the in-memory attribute values for each matched object. This keeps the ORM cache consistent with the database without requiring an explicit session.expire() call or a full cache flush.
How This Fails in Real Systems
A payments system processed billing changes using low-level bulk updates via SQLAlchemy Core for performance. Because the concurrent API instances held long-running sessions, they used stale ORM model instances, charging users the old subscription rates for up to 12 hours before a service restart cleared the cache.