python - Django 从原始查询创建管理列表

Question

我愿意在管理员中创建查询集列表模板

查询集是通过 cursor.execute() 获得的，因此可能需要一个假模型。

基本上我只想利用 Django admin list 提供的分页和过滤功能。

这是我的代码的相关部分

模型.py

class Query(object):

   def __init__(self, sql):

       self.sql = sql


   def execute_query(self):

        cursor = connection.cursor()
        start_time = time()

        try:
            cursor.execute(self.sql)
        except DatabaseError as e:
            cursor.close()
            raise e

        return cursor, ((time() - start_time) * 1000)

admin.py

 class QueryAdmin(admin.ModelAdmin):
 ....


admin.site.register(Query, QueryAdmin)

Answer 1

我刚刚实现了这个功能。您需要新的 View RawChangeList 扩展 ChangeList 来自 Django 的 View :

from django.contrib.admin.views.main import ChangeList
from django.db import connections

class RawChangeList(ChangeList):
    """
    Extended Django ChangeList to be able show data from RawQueryset.
    """
    def get_count(self):
        connection = connections[self.queryset.db]
        with connection.cursor() as c:
            if connection.vendor == 'microsoft':  # CTE in subquery is not working in SQL Server
                c.execute(self.queryset.raw_query)
                c.execute('SELECT @@ROWCOUNT')
            else:
                query = 'SELECT COUNT(*) FROM ({query}) AS sq'
                c.execute(query.format(query=self.queryset.raw_query))

            return c.fetchone()[0]

    def get_queryset_slice(self):
        connection = connections[self.queryset.db]
        if connection.vendor == 'microsoft':
            # SQL Server needs ordered query for slicing
            if hasattr(self.queryset, 'ordered') and self.queryset.ordered:
                query = '{query}'
            else:
                query = '{query} ORDER BY 1'
            query += ' OFFSET {offset} ROWS FETCH NEXT {limit} ROWS ONLY'
        else:
            query = '{query} LIMIT {limit} OFFSET {offset}'

        return self.queryset.model.objects.raw(
            query.format(
                query=self.queryset.raw_query,
                offset=self.page_num * self.list_per_page,
                limit=(self.page_num + 1) * self.list_per_page - self.page_num * self.list_per_page,
            )
        )

    def get_queryset(self, request):
        """
        Overriding to avoid applying filters in ChangeList because RawQueryset has not filter method.
        So any filters has to be applied manually for now.
        """
        qs = self.root_queryset
        if not hasattr(qs, 'count'):
            qs.count = lambda: self.get_count()
        return qs

    def get_results(self, request):
        if self.show_all:
            qs = self.queryset
        else:
            qs = self.get_queryset_slice()

        paginator = self.model_admin.get_paginator(request, self.queryset, self.list_per_page)

        self.result_count = paginator.count
        self.show_full_result_count = False
        self.show_admin_actions = True
        self.full_result_count = 0
        self.result_list = list(qs)
        self.can_show_all = True
        self.multi_page = True
        self.paginator = paginator

然后在管理中使用这个RawChangeList:

@admin.register(MyModel)
class MyModelAdmin(admin.ModelAdmin):

    def get_changelist(self, request, **kwargs):
        return RawChangeList

    def get_queryset(self, request):
        return MyModel.objects.raw('SELECT * FROM my_app_my_model')

    def get_object(self, request, object_id, from_field=None):
        return MyModel.objects.raw('SELECT * FROM my_app_my_model WHERE id=%s', (object_id, ))[0]