我愿意在管理员中创建查询集列表模板
查询集是通过 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)
请您参考如下方法:
我刚刚实现了这个功能。您需要新的 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]
