"""This series of tests illustrates different ways to INSERT a large numberof rows in bulk."""fromsqlalchemyimportbindparamfromsqlalchemyimportColumnfromsqlalchemyimportcreate_enginefromsqlalchemyimportIntegerfromsqlalchemyimportStringfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportSessionfrom.importProfilerBase=declarative_base()engine=NoneclassCustomer(Base):__tablename__="customer"id=Column(Integer,primary_key=True)name=Column(String(255))description=Column(String(255))Profiler.init("bulk_inserts",num=100000)@Profiler.setupdefsetup_database(dburl,echo,num):globalengineengine=create_engine(dburl,echo=echo)Base.metadata.drop_all(engine)Base.metadata.create_all(engine)@Profiler.profiledeftest_flush_no_pk(n):"""Individual INSERT statements via the ORM, calling upon last row id"""session=Session(bind=engine)forchunkinrange(0,n,1000):session.add_all([Customer(name="customer name %d"%i,description="customer description %d"%i,)foriinrange(chunk,chunk+1000)])session.flush()session.commit()@Profiler.profiledeftest_bulk_save_return_pks(n):"""Individual INSERT statements in "bulk", but calling upon last row id"""session=Session(bind=engine)session.bulk_save_objects([Customer(name="customer name %d"%i,description="customer description %d"%i,)foriinrange(n)],return_defaults=True,)session.commit()@Profiler.profiledeftest_flush_pk_given(n):"""Batched INSERT statements via the ORM, PKs already defined"""session=Session(bind=engine)forchunkinrange(0,n,1000):session.add_all([Customer(id=i+1,name="customer name %d"%i,description="customer description %d"%i,)foriinrange(chunk,chunk+1000)])session.flush()session.commit()@Profiler.profiledeftest_bulk_save(n):"""Batched INSERT statements via the ORM in "bulk", discarding PKs."""session=Session(bind=engine)session.bulk_save_objects([Customer(name="customer name %d"%i,description="customer description %d"%i,)foriinrange(n)])session.commit()@Profiler.profiledeftest_bulk_insert_mappings(n):"""Batched INSERT statements via the ORM "bulk", using dictionaries."""session=Session(bind=engine)session.bulk_insert_mappings(Customer,[dict(name="customer name %d"%i,description="customer description %d"%i,)foriinrange(n)],)session.commit()@Profiler.profiledeftest_core_insert(n):"""A single Core INSERT construct inserting mappings in bulk."""conn=engine.connect()conn.execute(Customer.__table__.insert(),[dict(name="customer name %d"%i,description="customer description %d"%i,)foriinrange(n)],)@Profiler.profiledeftest_dbapi_raw(n):"""The DBAPI's API inserting rows in bulk."""conn=engine.pool._creator()cursor=conn.cursor()compiled=(Customer.__table__.insert().values(name=bindparam("name"),description=bindparam("description")).compile(dialect=engine.dialect))ifcompiled.positional:args=(("customer name %d"%i,"customer description %d"%i)foriinrange(n))else:args=(dict(name="customer name %d"%i,description="customer description %d"%i,)foriinrange(n))cursor.executemany(str(compiled),list(args))conn.commit()conn.close()if__name__=="__main__":Profiler.main()