Source code for examples.performance.single_inserts
"""In this series of tests, we're looking at a method that inserts a rowwithin a distinct transaction, and afterwards returns to essentially a"closed" state. This would be analogous to an API call that starts upa database connection, inserts the row, commits and closes."""fromsqlalchemyimportbindparamfromsqlalchemyimportColumnfromsqlalchemyimportcreate_enginefromsqlalchemyimportIntegerfromsqlalchemyimportpoolfromsqlalchemyimportStringfromsqlalchemy.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("single_inserts",num=10000)@Profiler.setupdefsetup_database(dburl,echo,num):globalengineengine=create_engine(dburl,echo=echo)ifengine.dialect.name=="sqlite":engine.pool=pool.StaticPool(creator=engine.pool._creator)Base.metadata.drop_all(engine)Base.metadata.create_all(engine)@Profiler.profiledeftest_orm_commit(n):"""Individual INSERT/COMMIT pairs via the ORM"""foriinrange(n):session=Session(bind=engine)session.add(Customer(name="customer name %d"%i,description="customer description %d"%i,))session.commit()@Profiler.profiledeftest_bulk_save(n):"""Individual INSERT/COMMIT pairs using the "bulk" API """foriinrange(n):session=Session(bind=engine)session.bulk_save_objects([Customer(name="customer name %d"%i,description="customer description %d"%i,)])session.commit()@Profiler.profiledeftest_bulk_insert_dictionaries(n):"""Individual INSERT/COMMIT pairs using the "bulk" API with dictionaries"""foriinrange(n):session=Session(bind=engine)session.bulk_insert_mappings(Customer,[dict(name="customer name %d"%i,description="customer description %d"%i,)],)session.commit()@Profiler.profiledeftest_core(n):"""Individual INSERT/COMMIT pairs using Core."""foriinrange(n):withengine.begin()asconn:conn.execute(Customer.__table__.insert(),dict(name="customer name %d"%i,description="customer description %d"%i,),)@Profiler.profiledeftest_core_query_caching(n):"""Individual INSERT/COMMIT pairs using Core with query caching"""cache={}ins=Customer.__table__.insert()foriinrange(n):withengine.begin()asconn:conn.execution_options(compiled_cache=cache).execute(ins,dict(name="customer name %d"%i,description="customer description %d"%i,),)@Profiler.profiledeftest_dbapi_raw_w_connect(n):"""Individual INSERT/COMMIT pairs w/ DBAPI + connection each time"""_test_dbapi_raw(n,True)@Profiler.profiledeftest_dbapi_raw_w_pool(n):"""Individual INSERT/COMMIT pairs w/ DBAPI + connection pool"""_test_dbapi_raw(n,False)def_test_dbapi_raw(n,connect):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))sql=str(compiled)ifconnect:forarginargs:# there's no connection pool, so if these were distinct# calls, we'd be connecting each timeconn=engine.pool._creator()cursor=conn.cursor()cursor.execute(sql,arg)cursor.lastrowidconn.commit()conn.close()else:forarginargs:conn=engine.raw_connection()cursor=conn.cursor()cursor.execute(sql,arg)cursor.lastrowidconn.commit()conn.close()if__name__=="__main__":Profiler.main()