Лучший способ разбить необработанный SQL-запрос в представлении Django REST ListAPI?
У меня есть сырой SQL-запрос, который я использую для создания набора запросов для представления REST ListAPI Django. Это соответствует следующему (прошу прощения за бессмысленные имена):
class MyView(ListAPIView):
serializer_class = MySerializer
paginate_by = 10
def get_queryset(self):
params = {
"uid": str(self.request.user.id),
"param": str(self.kwargs['param'])
}
query = 'SELECT f.id ' \
'FROM myapp_foo f, myapp_bar b ' \
'WHERE b.foo_id = f.id AND ' \
'b.param >= %(param)s AND ' \
'f.dt_tm >= NOW() AND ' \
'(SELECT COUNT(*) FROM myapp_baz z ' \
'WHERE z.user_id = %(uid)s AND ' \
'z.qux_id = f.qux_id) = 0 ' \
'ORDER BY f.dt_tm;'
return Foo.objects.raw(query, params)
Это дает ошибку:
object of type 'RawQuerySet' has no len()
Я хотел бы рассчитать количество с аналогичным запросом SQL, а затем использовать параметры LIMIT и OFFSET, чтобы сделать нумерацию страниц. Я прочитал несколько предложений, в которых элементы списка учитываются для получения len, но это не кажется удовлетворительным, поскольку это было бы неэффективно, если бы в запросе не было небольшого LIMIT (что в любом случае не позволило бы использовать нумерацию страниц).
Обновление: я только заметил, что paginate_by ожидает устаревания.
Для начала как добавить метод count к возвращаемому объекту?
4 ответа
Более эффективным решением, чем другие альтернативы, было бы написать свой собственный RawQuerySet замена. Я показываю код ниже, но вы также можете получить доступ к нему здесь. Абсолютно не гарантируется отсутствие ошибок; тем не менее я использую его в Django 1.11 на Python 3 (с PostgreSQL в качестве базы данных; должен работать и с MySQL). Проще говоря, этот класс добавляет соответствующий LIMIT а также OFFSET предложения к вашему сырому SQL-запросу. В этом нет ничего сумасшедшего, просто простая конкатенация строк, поэтому не включайте эти пункты в ваш необработанный SQL-запрос.
Класс
from django.db import models
from django.db.models import sql
from django.db.models.query import RawQuerySet
class PaginatedRawQuerySet(RawQuerySet):
def __init__(self, raw_query, **kwargs):
super(PaginatedRawQuerySet, self).__init__(raw_query, **kwargs)
self.original_raw_query = raw_query
self._result_cache = None
def __getitem__(self, k):
"""
Retrieves an item or slice from the set of results.
"""
if not isinstance(k, (slice, int,)):
raise TypeError
assert ((not isinstance(k, slice) and (k >= 0)) or
(isinstance(k, slice) and (k.start is None or k.start >= 0) and
(k.stop is None or k.stop >= 0))), \
"Negative indexing is not supported."
if self._result_cache is not None:
return self._result_cache[k]
if isinstance(k, slice):
qs = self._clone()
if k.start is not None:
start = int(k.start)
else:
start = None
if k.stop is not None:
stop = int(k.stop)
else:
stop = None
qs.set_limits(start, stop)
return qs
qs = self._clone()
qs.set_limits(k, k + 1)
return list(qs)[0]
def __iter__(self):
self._fetch_all()
return iter(self._result_cache)
def count(self):
if self._result_cache is not None:
return len(self._result_cache)
return self.model.objects.count()
def set_limits(self, start, stop):
limit_offset = ''
new_params = tuple()
if start is None:
start = 0
elif start > 0:
new_params += (start,)
limit_offset = ' OFFSET %s'
if stop is not None:
new_params = (stop - start,) + new_params
limit_offset = 'LIMIT %s' + limit_offset
self.params = self.params + new_params
self.raw_query = self.original_raw_query + limit_offset
self.query = sql.RawQuery(sql=self.raw_query, using=self.db, params=self.params)
def _fetch_all(self):
if self._result_cache is None:
self._result_cache = list(super().__iter__())
def __repr__(self):
return '<%s: %s>' % (self.__class__.__name__, self.model.__name__)
def __len__(self):
self._fetch_all()
return len(self._result_cache)
def _clone(self):
clone = self.__class__(raw_query=self.raw_query, model=self.model, using=self._db, hints=self._hints,
query=self.query, params=self.params, translations=self.translations)
return clone
Как это использовать
Пользовательский менеджер
Я использую запрос выше через пользовательский менеджер:
class MyModelRawManager(models.Manager):
def raw(self, raw_query, params=None, translations=None, using=None):
if using is None:
using = self.db
return PaginatedRawQuerySet(raw_query, model=self.model, params=params, translations=translations, using=using)
def my_raw_sql_method(some_arg):
# set up your query and params
query = 'your query'
params = ('your', 'params', 'tuple')
return self.raw(raw_query=query, params=params)
Пользовательский класс нумерации страниц
Для завершения я также включаю класс нумерации страниц:
from rest_framework.pagination import PageNumberPagination
class MyModelResultsPagination(PageNumberPagination):
"""Fixed page-size pagination with 10 items."""
page_size = 10
max_page_size = 10
Ваш список APIView
class MyModelView(generics.ListAPIView):
serializer_class = MyModelSerializer
pagination_class = MyModelResultsPagination
def get_queryset(self):
return MyModel.raw_manager.my_raw_sql_method(some_arg)
Слово предостережения
PaginatedRawQuerySet Класс, хотя он и был для меня функциональным, не был тщательно протестирован, но я полагаю, что он дает представление о том, что требуется для более эффективного решения, чем выбор всех элементов в вашем наборе запросов для каждого вызова.
Вы можете заметить, что есть обычай count реализация метода (изначально отсутствует в RawQuerySet), который рассчитывается по телефону self.model.objects.count(), Без этого метода paginator будет оценивать len(your_raw_queryset), что будет иметь такое же влияние на производительность, как и другой ответ.
Этот класс не является универсальной заменой для RawQuerySet, что означает, что вы должны добавить свои собственные настройки, чтобы они соответствовали вашим потребностям.
Например, если вам нужно что-то более сложное, вы можете добавить еще один атрибут к PaginatedRawQuerySet класс, называется raw_count_query, который затем будет вызван внутри count() вместо подсчета всех объектов, как это происходит сейчас (это будет использоваться в случаях, когда вам нужна фильтрация; raw_count_query предоставит SQL для подсчета подмножества на основе ваших условий).
Если вы приведете необработанный набор запросов к списку перед его возвратом, это должно предотвратить 'RawQuerySet' has no len() ошибка.
return list(Foo.objects.raw(query))
Как вы говорите, это будет неэффективно, так как будет загружать весь набор запросов.
Можно было бы написать собственный класс разбиения на страницы, который эффективно разбивает на страницы с использованием limit и offset, и использовать его в своем представлении с атрибутом pagination_class.
У меня была та же проблема, и я обнаружил, что вместо использования raw вы можете использовать extra:
(...)
return Foo.objects.extra(where=query, params=params)
дополнительные переменные
where=['data->>"$.SOMETHING" = %s OR data->>"$.SOMETHING" = %s OR data->>"$.SOMETHING" = %s', 'data->>"$.GROUP" LIKE %s']
params=['EX1', 'EX2', 'EX3', '%EXEMPLE4%']
Примечание: основной вопрос заключается в использовании RawQuerySet с такими же свойствами QuerySet, лучший способ IMHO - использовать дополнительный API QuerySet, если это возможно.
Улучшенное решение от:
Класс
Я изменил содержаниеcountметод, определенный вPaginatedRawQuerySetclass (фрагмент кода взят отсюда ) и небольшое исправление методаset_limits, такие как:
def count(self):
from django.db import connection
if self._result_cache is not None:
return len(self._result_cache)
params = ["""'%s'""" % p for p in self.params]
sql = "SELECT COUNT(*) FROM (" + (self.raw_query % tuple(params)) + ") B;"
cursor = connection.cursor()
cursor.execute(sql)
row = cursor.fetchone()
return row[0]
def set_limits(self, start, stop):
limit_offset = ''
new_params = tuple()
if start is None:
start = 0
elif start > 0:
new_params += (start,)
limit_offset = ' OFFSET %s'
if stop is not None:
new_params = (stop - start,) + new_params
limit_offset = 'LIMIT %s' + limit_offset
# here is important to convert 'self.params' to a tuple.
self.params = tuple(self.params or ()) + new_params
self.raw_query = self.original_raw_query + limit_offset
self.query = sql.RawQuery(sql=self.raw_query, using=self.db, params=self.params)
Как это использовать
Пользовательский менеджер
class MyModelRawManager(models.Manager):
def raw(self, raw_query, params=None, translations=None, using=None):
if using is None:
using = self.db
return PaginatedRawQuerySet(raw_query, model=self.model, params=params, translations=translations, using=using)
Пользовательский класс нумерации страниц
from rest_framework.pagination import PageNumberPagination
class MyModelResultsPagination(PageNumberPagination):
"""Fixed page-size pagination with 10 items."""
page_size = 10
max_page_size = 10
Модель
class ModelTest(models.Model):
raw_manager = MyModelRawManager()
Вид
class MyModelView(generics.ListAPIView):
serializer_class = MyModelSerializer
pagination_class = MyModelResultsPagination
def get_queryset(self):
query_sql = "select * from offices"
return ModelTest.raw_manager.raw_query(query_sql)
def list(self, request, *args, **kwargs):
self.queryset = self.get_queryset()
paginated = self.paginate_queryset(self.queryset)
data = self.serializer_class(paginated, many=True).data
response = self.paginator.get_paginated_response(data)
return response