Source code for examples.performance.short_selects
"""This series of tests illustrates different ways to SELECT a singlerecord by primary key"""importrandomfromsqlalchemyimportbindparamfromsqlalchemyimportColumnfromsqlalchemyimportcreate_enginefromsqlalchemyimportIdentityfromsqlalchemyimportIntegerfromsqlalchemyimportselectfromsqlalchemyimportStringfromsqlalchemy.extimportbakedfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.futureimportselectasfuture_selectfromsqlalchemy.ormimportdeferredfromsqlalchemy.ormimportSessionfromsqlalchemy.sqlimportlambdasfrom.importProfilerBase=declarative_base()engine=Noneids=range(1,11000)classCustomer(Base):__tablename__="customer"id=Column(Integer,Identity(),primary_key=True)name=Column(String(255))description=Column(String(255))q=Column(Integer)p=Column(Integer)x=deferred(Column(Integer))y=deferred(Column(Integer))z=deferred(Column(Integer))Profiler.init("short_selects",num=10000)@Profiler.setupdefsetup_database(dburl,echo,num):globalengineengine=create_engine(dburl,echo=echo)Base.metadata.drop_all(engine)Base.metadata.create_all(engine)sess=Session(engine)sess.add_all([Customer(id=i,name="c%d"%i,description="c%d"%i,q=i*10,p=i*20,x=i*30,y=i*40,)foriinids])sess.commit()@Profiler.profiledeftest_orm_query_classic_style(n):"""classic ORM query of the full entity."""session=Session(bind=engine)forid_inrandom.sample(ids,n):session.query(Customer).filter(Customer.id==id_).one()@Profiler.profiledeftest_orm_query_new_style(n):"""new style ORM select() of the full entity."""session=Session(bind=engine)forid_inrandom.sample(ids,n):stmt=future_select(Customer).where(Customer.id==id_)session.execute(stmt).scalar_one()@Profiler.profiledeftest_orm_query_new_style_using_embedded_lambdas(n):"""new style ORM select() of the full entity w/ embedded lambdas."""session=Session(bind=engine)forid_inrandom.sample(ids,n):stmt=future_select(lambda:Customer).where(lambda:Customer.id==id_)session.execute(stmt).scalar_one()@Profiler.profiledeftest_orm_query_new_style_using_external_lambdas(n):"""new style ORM select() of the full entity w/ external lambdas."""session=Session(bind=engine)forid_inrandom.sample(ids,n):stmt=lambdas.lambda_stmt(lambda:future_select(Customer))stmt+=lambdas:s.where(Customer.id==id_)session.execute(stmt).scalar_one()@Profiler.profiledeftest_orm_query_classic_style_cols_only(n):"""classic ORM query against columns"""session=Session(bind=engine)forid_inrandom.sample(ids,n):session.query(Customer.id,Customer.name,Customer.description).filter(Customer.id==id_).one()@Profiler.profiledeftest_orm_query_new_style_ext_lambdas_cols_only(n):"""new style ORM query w/ external lambdas against columns."""s=Session(bind=engine)forid_inrandom.sample(ids,n):stmt=lambdas.lambda_stmt(lambda:future_select(Customer.id,Customer.name,Customer.description))+(lambdas:s.filter(Customer.id==id_))s.execute(stmt).one()@Profiler.profiledeftest_baked_query(n):"""test a baked query of the full entity."""bakery=baked.bakery()s=Session(bind=engine)forid_inrandom.sample(ids,n):q=bakery(lambdas:s.query(Customer))q+=lambdaq:q.filter(Customer.id==bindparam("id"))q(s).params(id=id_).one()@Profiler.profiledeftest_baked_query_cols_only(n):"""test a baked query of only the entity columns."""bakery=baked.bakery()s=Session(bind=engine)forid_inrandom.sample(ids,n):q=bakery(lambdas:s.query(Customer.id,Customer.name,Customer.description))q+=lambdaq:q.filter(Customer.id==bindparam("id"))q(s).params(id=id_).one()@Profiler.profiledeftest_core_new_stmt_each_time(n):"""test core, creating a new statement each time."""withengine.connect()asconn:forid_inrandom.sample(ids,n):stmt=select(Customer.__table__).where(Customer.id==id_)row=conn.execute(stmt).first()tuple(row)@Profiler.profiledeftest_core_new_stmt_each_time_compiled_cache(n):"""test core, creating a new statement each time, but using the cache."""compiled_cache={}withengine.connect().execution_options(compiled_cache=compiled_cache)asconn:forid_inrandom.sample(ids,n):stmt=select(Customer.__table__).where(Customer.id==id_)row=conn.execute(stmt).first()tuple(row)@Profiler.profiledeftest_core_reuse_stmt(n):"""test core, reusing the same statement (but recompiling each time)."""stmt=select(Customer.__table__).where(Customer.id==bindparam("id"))withengine.connect()asconn:forid_inrandom.sample(ids,n):row=conn.execute(stmt,{"id":id_}).first()tuple(row)@Profiler.profiledeftest_core_reuse_stmt_compiled_cache(n):"""test core, reusing the same statement + compiled cache."""stmt=select(Customer.__table__).where(Customer.id==bindparam("id"))compiled_cache={}withengine.connect().execution_options(compiled_cache=compiled_cache)asconn:forid_inrandom.sample(ids,n):row=conn.execute(stmt,{"id":id_}).first()tuple(row)if__name__=="__main__":Profiler.main()