Anton Shestakov <engored@ya.ru>, Sat, 29 Nov 2014 12:56:58 +0800
find-closest: searching for a thing in parent dirs, a la jQuery.closest
qmonitor/example.py
Permissions: -rw-r--r--
from collections import Counter from contextlib import contextmanager from sqlalchemy import create_engine from sqlalchemy import event from sqlalchemy import Column, ForeignKey, Integer, String from sqlalchemy.orm import relationship, backref, joinedload from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base engine = create_engine('sqlite:///:memory:') Session = sessionmaker(bind=engine) Base = declarative_base() id = Column(Integer, primary_key=True) fullname = Column(String) password = Column(String) __tablename__ = 'addresses' id = Column(Integer, primary_key=True) email_address = Column(String, nullable=False) user_id = Column(Integer, ForeignKey('users.id')) user = relationship("User", backref=backref('addresses', order_by=id)) Base.metadata.create_all(engine) User(name='wendy', fullname='Wendy Williams', password='foobar'), User(name='mary', fullname='Mary Contrary', password='xxg527'), User(name='fred', fullname='Fred Flinstone', password='blah') Address(email_address=user.name + '@example.com'), Address(email_address=user.name + '@work.corp.com') def __init__(self, max_queries): self.max_queries = max_queries """ User requires assistance in query minimization. qc = Counter(zip(*self.queries)[0]) if qc.values().count(1) < len(qc): '** Repeated {0} times: {1}'.format(c, q) print (u'Too many queries, try eliminating duplicates:\n' + print u'Too many queries:\n' + u'\n'.join(zip(*self.queries)[0]) """ Check saved database queries. if len(self.queries) > self.max_queries: def qmonitor(max_queries=15): qm = QMonitor(max_queries) def before_cursor_execute(conn, cursor, statement, parameters, context, executemany): context._query_start_time = time.time() def after_cursor_execute(conn, cursor, statement, parameters, context, executemany): total = time.time() - context._query_start_time qm.queries.append((statement, total)) event.listen(engine, 'before_cursor_execute', before_cursor_execute) event.listen(engine, 'after_cursor_execute', after_cursor_execute) event.remove(engine, 'before_cursor_execute', before_cursor_execute) event.remove(engine, 'after_cursor_execute', after_cursor_execute) q = session.query(Address) with qmonitor(max_queries=1) as qm: print address.email_address + ' - ' + address.user.fullname print '== Optimized query' q = session.query(Address).options(joinedload('user')) with qmonitor(max_queries=1) as qm: print address.email_address + ' - ' + address.user.fullname