Estrategias de actualización masiva de SQLAlchemy

Básicamente, lo que está haciendo es omitir el ORM para optimizar el rendimiento. Por lo tanto, no se sorprenda de que esté "replicando el trabajo que está haciendo el ORM" porque eso es exactamente lo que necesita hacer.

A menos que tenga muchos lugares donde necesite hacer actualizaciones masivas como esta, no recomendaría el enfoque de eventos mágicos; simplemente escribir las consultas explícitas es mucho más sencillo.

Lo que recomiendo hacer es usar SQLAlchemy Core en lugar del ORM para hacer la actualización:

ledger = Table("ledger", db.metadata,
    Column("wallet_id", Integer, primary_key=True),
    Column("new_balance", Float),
    prefixes=["TEMPORARY"],
)


wallets = db_session.query(Wallet).all()

# figure out new balances
balance_map = {}
for w in wallets:
    balance_map[w.id] = calculate_new_balance(w)

# create temp table with balances we need to update
ledger.create(bind=db.session.get_bind())

# insert update data
db.session.execute(ledger.insert().values([{"wallet_id": k, "new_balance": v}
                                           for k, v in balance_map.items()])

# perform update
db.session.execute(Wallet.__table__
                         .update()
                         .values(balance=ledger.c.new_balance)
                         .where(Wallet.__table__.c.id == ledger.c.wallet_id))

# drop temp table
ledger.drop(bind=db.session.get_bind())

# commit changes
db.session.commit()