8000 Profiling · sqlalchemy/sqlalchemy Wiki · GitHub
[go: up one dir, main page]
Skip to content

Profiling

Jump to bottom
mike bayer edited this page Jun 16, 2022 · 6 revisions

Query Timing

from sqlalchemy import event
from sqlalchemy.engine import Engine
import time
import logging
 
logging.basicConfig()
logger = logging.getLogger("myapp.sqltime")
logger.setLevel(logging.DEBUG)

@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement,
                        parameters, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())
    logger.debug("Start Query: %s" % statement)

@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement,
                        parameters, context, executemany):
    total = time.time() - conn.info['query_start_time'].pop(-1)
    logger.debug("Query Complete!")
    logger.debug("Total Time: %f" % total)

if __name__ == '__main__':
    from sqlalchemy import *
 
    engine = create_engine('sqlite://')
 
    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))
    )

Note that to apply the events to just a particular engine, not the Engine class as a whole, use event.listen() with the engine:

engine = create_engine('sqlite://')
event.listen(engine, "before_cursor_execute", before_cursor_execute)
event.listen(engine, "after_cursor_execute", after_cursor_execute)
Clone this wiki locally
0