SQLAlchemyQueryTimer.py

"""See http://techspot.zzzeek.org/?p=31"""

from sqlalchemy.interfaces import ConnectionProxy
import time
import logging
 
logging.basicConfig()
logger = logging.getLogger("myapp.sqltime")
logger.setLevel(logging.DEBUG)
 
class TimerProxy(ConnectionProxy):
    def cursor_execute(self, execute, cursor, statement, parameters, context, executemany):
        now = time.time()
        try:
            return execute(cursor, statement, parameters, context)
        finally:
            total = time.time() - now
            logger.debug("Query: %s" % statement)
            logger.debug("Total Time: %f" % total)
 
if __name__ == '__main__':
    from sqlalchemy import *
 
    engine= create_engine('sqlite://', proxy=TimerProxy())
 
    m1 = MetaData(engine)
    t1 = Table("sometable", m1, 
            Column("id", Integer, primary_key=True),
            Column("data", String(255), nullable=False),
        )
 
    conn = engine.connect()
    m1.create_all(conn)
 
    conn.execute(
        t1.insert(), 
        [{"data":"entry %d" % x} for x in xrange(100000)]
    )
 
    conn.execute(
        t1.select().where(t1.c.data.between("entry 25", "entry 7800")).order_by(desc(t1.c.data))
    )