poolload.py

来自「SQLAlchemy. 经典的Python ORM框架。学习必看。」· Python 代码 · 共 39 行

# load test of connection pool
import testenv; testenv.configure_for_tests()
import thread, time
from sqlalchemy import *
import sqlalchemy.pool as pool
from testlib import testing

db = create_engine(testing.db.url, pool_timeout=30, echo_pool=True)
metadata = MetaData(db)

users_table = Table('users', metadata,
  Column('user_id', Integer, primary_key=True),
  Column('user_name', String(40)),
  Column('password', String(10)))
metadata.drop_all()
metadata.create_all()

users_table.insert().execute([{'user_name':'user#%d' % i, 'password':'pw#%d' % i} for i in range(1000)])

def runfast():
    while True:
        c = db.pool.connect()
        time.sleep(.5)
        c.close()
#        result = users_table.select(limit=100).execute()
#        d = {}
#        for row in result:
#            for col in row.keys():
#                d[col] = row[col]
#        time.sleep(.005)
#        result.close()
        print "runfast cycle complete"

#thread.start_new_thread(runslow, ())
for x in xrange(0,50):
    thread.start_new_thread(runfast, ())

time.sleep(100)