Release: 1.1.0b1 | Release Date: not released

SQLAlchemy 1.1 Documentation

Source code for examples.performance.bulk_updates

"""This series of tests illustrates different ways to UPDATE a large number
of rows in bulk.

from . import Profiler

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, create_engine, bindparam
from sqlalchemy.orm import Session

Base = declarative_base()
engine = None

class Customer(Base):
    __tablename__ = "customer"
    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    description = Column(String(255))

Profiler.init("bulk_updates", num=100000)

def setup_database(dburl, echo, num):
    global engine
    engine = create_engine(dburl, echo=echo)

    s = Session(engine)
    for chunk in range(0, num, 10000):
        s.bulk_insert_mappings(Customer, [
                'name': 'customer name %d' % i,
                'description': 'customer description %d' % i
            } for i in range(chunk, chunk + 10000)

def test_orm_flush(n):
    """UPDATE statements via the ORM flush process."""
    session = Session(bind=engine)
    for chunk in range(0, n, 1000):
        customers = session.query(Customer).\
            filter(, chunk + 1000)).all()
        for customer in customers:
            customer.description += "updated"