Demo of how to test for n+1 queries produced by sqlalchemy

sqlalchemy_most_frequent_queries.py

# for use with sqlalchemy

def test_foo():
  stmts = []
  def catch_queries(conn, cursor, statement, a, b, c):
    stmts.append(statement)
  event.listen(engine, "before_cursor_execute", catch_queries)
  do_thing()
  assert _most_frequent_queries(queries=stmts, query_count=4, query_type='SELECT', query_print_length=50) == [('SELECT X FROM Y', 5)]

def _most_frequent_queries(queries, query_count, query_type, query_print_length):
    select_sql_statement_groupings = {}
    for statement in queries:
        if statement.startswith(query_type):
            select_sql_statement_groupings.setdefault(statement, 0)
            select_sql_statement_groupings[statement] += 1
    return list(map(lambda x: (x[0][:query_print_length], x[1]), list(OrderedDict(
        sorted(select_sql_statement_groupings.items(), key=lambda x: x[1], reverse=True)).items())[:query_count]))