from__future__importannotationsfromsqlalchemyimportbindparamfromsqlalchemyimportColumnfromsqlalchemyimportcreate_enginefromsqlalchemyimportIdentityfromsqlalchemyimportinsertfromsqlalchemyimportIntegerfromsqlalchemyimportStringfromsqlalchemy.ormimportdeclarative_basefromsqlalchemy.ormimportSessionfrom.importProfiler"""This series of tests illustrates different ways to INSERT a large numberof rows in bulk."""Base=declarative_base()classCustomer(Base):__tablename__="customer"id=Column(Integer,Identity(),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):"""INSERT statements via the ORM (batched with RETURNING if available), fetching generated 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_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_orm_bulk_insert(n):"""Batched INSERT statements via the ORM in "bulk", not returning rows"""session=Session(bind=engine)session.execute(insert(Customer),[{"name":"customer name %d"%i,"description":"customer description %d"%i,}foriinrange(n)],)session.commit()@Profiler.profiledeftest_orm_insert_returning(n):"""Batched INSERT statements via the ORM in "bulk", returning new Customer objects"""session=Session(bind=engine)customer_result=session.scalars(insert(Customer).returning(Customer),[{"name":"customer name %d"%i,"description":"customer description %d"%i,}foriinrange(n)],)# this step is where the rows actually become objectscustomers=customer_result.all()# noqa: F841session.commit()@Profiler.profiledeftest_core_insert(n):"""A single Core INSERT construct inserting mappings in bulk."""withengine.begin()asconn: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()