Как подсчитать запросы sqlalchemy в модульных тестах

В Django я часто утверждаю количество запросов, которые должны быть сделаны, чтобы модульные тесты ломали новые проблемы с запросом N + 1

from django import db
from django.conf import settings
settings.DEBUG=True

class SendData(TestCase):
    def test_send(self):
        db.connection.queries = []
        event = Events.objects.all()[1:]
        s = str(event) # QuerySet is lazy, force retrieval
        self.assertEquals(len(db.connection.queries), 2)

В SQLAlchemy трассировка на STDOUT включена, установив флаг echo двигатель

engine.echo=True

Каков наилучший способ написать тесты, которые подсчитывают количество запросов, сделанных SQLAlchemy?

class SendData(TestCase):
    def test_send(self):
        event = session.query(Events).first()
        s = str(event)
        self.assertEquals( ... , 2)

Ответы

Ответ 1

Использовать SQLAlchemy Основные события для выполнения запросов регистрации/отслеживания (вы можете прикрепить их к своим модуляционным тестам, чтобы они не влияли на вашу производительность на самом деле:

event.listen(engine, "before_cursor_execute", catch_queries)

Теперь вы пишете функцию catch_queries, где путь зависит от того, как вы тестируете. Например, вы можете определить эту функцию в своем тестовом заявлении:

def test_something(self):
    stmts = []
    def catch_queries(conn, cursor, statement, ...):
        stmts.append(statement)
    # Now attach it as a listener and work with the collected events after running your test

Вышеупомянутый метод - просто вдохновение. Для расширенных случаев вы, вероятно, хотели бы иметь глобальный кеш событий, которые вы опорожняете после каждого теста. Причина в том, что до 0,9 (текущий dev) нет API для удаления прослушивателей событий. Таким образом, один глобальный слушатель получает доступ к глобальному списку.

Ответ 2

Я создал для этой цели класс менеджера контекста:

class DBStatementCounter(object):
    """
    Use as a context manager to count the number of execute() performed
    against the given sqlalchemy connection.

    Usage:
        with DBStatementCounter(conn) as ctr:
            conn.execute("SELECT 1")
            conn.execute("SELECT 1")
        assert ctr.get_count() == 2
    """
    def __init__(self, conn):
        self.conn = conn
        self.count = 0
        # Will have to rely on this since sqlalchemy 0.8 does not support
        # removing event listeners
        self.do_count = False
        sqlalchemy.event.listen(conn, 'after_execute', self.callback)

    def __enter__(self):
        self.do_count = True
        return self

    def __exit__(self, *_):
        self.do_count = False

    def get_count(self):
        return self.count

    def callback(self, *_):
        if self.do_count:
            self.count += 1