一、问题
xadmin后台功能很强大，特别在导出的时候格式有xls/xlsx、csv、xml、json。实际常用的还是前面2种。xls格式使用的xlwt，有个缺陷，导出数据过大时，会报ValueError: row index was 65536, not allowed by .xls format ...,使用xlsxwriter导出为xlsx格式，做个分页导出，十几万都不在话下（亲测）。
在xadmin/templates/xadmin/blocks/model_list.top_toolbar.importexport.export.html源码中可以看到这5种格式：
后台导出按钮下拉只有4种：
，xlsx格式在源码中可以找到答案。
在xadmin/plugins/export.py中，
，只有安装xlsxwriter依赖才会显示
这个下拉按钮。
二、思路
xadmin目前是在点下载等待所有符合条件数据下载到内存再生成文件交给浏览器下载，如果下载数据太大就会拖死这个请求。所以在下载数据在内存过程中分段下载，前后端配合传参，将参数存到redis(django session是存数据库django_session表中，存session会报错Packet for query is too large，需要修改mysql参数，max_allowed_packet，默认是1兆，所以存储到redis)，下载完删除。页数自增，每次后端根据总数和每次下载数确定状态还需不需要下载，最后标记下载完成，生成文件响应。
三、实现效果图
四、前后端代码
4-1.xadmin/templates/xadmin/blocks/model_list.top_toolbar.importexport.export.html
{% load i18n %} {% trans "Export" %} {% for et in export_types %} {% endfor %}
4-2.xadmin/plugins/export.py
def _down(self,context): # session中存储export_id {'dic':{'export_id':[]}} session_dic = {'export_id':[]} # params_obj = {'export_id': export_id, # 'n': n, # 'ajax': ajax} from sms.channels import reids_db # if 'dic' in self.request.session: # session_dic = self.request.session['dic'] # else: # # 初始化dic # self.request.session['dic'] = session_dic = {} if not reids_db.get('dic'): reids_db.set('dic', {}, 24 * 3600) # redis取出dic值为byte类型 使用eval转为字典 session_dic = eval(reids_db.get('dic')) # 从request对象获取export_id if hasattr(self.request,'params_obj'): export_id = self.request.params_obj['export_id'] else: datas = self._get_datas(context) return datas,{} if export_id: if export_id in session_dic: # export_id 是否在redis中 global session_list session_list = session_dic[export_id] else: # 初始化export_id session_dic[export_id] = session_list = [] reids_db.set('dic', session_dic, 24 * 3600) # 初始化传给前端的数据 context_datas = {} if self.request.params_obj['ajax']: if self.request.params_obj['p'] == '0': datas = self._get_datas(context) else: datas = self._get_datas(context)[1:] session_list += datas session_dic[export_id] = session_list reids_db.set('dic', session_dic, 24 * 3600) # 对比redis中data与总数 1：下载完，0：未下载完 # if len(session_list) + len(datas) >= context['result_count']: if len(session_list) >= context['result_count']: context_datas['status'] = 1 else: context_datas['status'] = 0 # session_list += datas # 返回前端下载总数和当前进度 context_datas['total'] = context['result_count'] context_datas['len'] = len(session_list) return '', context_datas # 最后一次返回所有数据下载 # 不是ajax 删除redis中的export_id if export_id: datas = session_list session_dic[export_id] = '' reids_db.set('dic', session_dic, 24 * 3600) else: datas = self._get_datas(context) return datas,context_datas def get_xlsx_export(self, context): datas,context_datas = self._down(context) output = io.BytesIO() # export_header = ( # self.request.GET.get('export_xlsx_header', 'off') == 'on') model_name = self.opts.verbose_name book = xlsxwriter.Workbook(output) sheet = book.add_worksheet( u"%s %s" % (_(u'Sheet'), force_text(model_name))) styles = { 'datetime': book.add_format({ 'num_format': 'yyyy-mm-dd hh:mm:ss'}), 'date': book.add_format({ 'num_format': 'yyyy-mm-dd'}), 'time': book.add_format({ 'num_format': 'hh:mm:ss'}), 'header': book.add_format({ 'font': 'name Times New Roman', 'color': 'red', 'bold': 'on', 'num_format': '#,##0.00'}), 'default': book.add_format()} # if not export_header: # datas = datas[1:] for rowx, row in enumerate(datas): for colx, value in enumerate(row): if rowx == 0: cell_style = styles['header'] else: if isinstance(value, datetime.datetime): cell_style = styles['datetime'] elif isinstance(value, datetime.date): cell_style = styles['date'] elif isinstance(value, datetime.time): cell_style = styles['time'] else: cell_style = styles['default'] sheet.write(rowx, colx, value, cell_style) book.close() output.seek(0) return output.getvalue(),context_datas def get_response(self, response, context, *args, **kwargs): file_type = self.request.GET.get('export_type', 'csv') content,context_datas = getattr(self, 'get_%s_export' % file_type)(context) if 'status' in context_datas.keys(): response = HttpResponse(json.dumps(context_datas), content_type="application/json") else: response = HttpResponse( content_type="%s; charset=UTF-8" % self.export_mimes[file_type]) file_name = self.opts.verbose_name.replace(' ', '_') # response['Content-Disposition'] = ('attachment; filename=%s.%s' % ( # file_name, file_type)).encode('utf-8') # 修复导出时gunicorn报错ascii from urllib.parse import quote response["Content-Disposition"] = \ "attachment; " \ "filenane=%s.%s;" \ "filename*=UTF-8''%s.%s" %( quote(file_name),file_type, quote(file_name),file_type ) response.write(content) return response
4-3.自己应用下的adminx.py中要做大量数据导出的model
class SMSLogAdmin(ReadonlyAdmin): list_display = ['id', 'my_mobile', 'status', 'req_time', 'ret_time', 'account', 'my_tally', 'my_price'] list_filter = ['account', 'status', 'req_time',] @property def list_per_page(self): import re path = self.request.get_full_path() pattern_res = re.findall('ajax',path) export_id = self.request.GET.get('export_id') p = self.request.GET.get('p') ajax = self.request.GET.get('ajax') # 将前端传过来的参数放到request对象中 if not hasattr(self.request,'params_obj'): self.request.params_obj = { 'export_id': export_id, 'p': p, 'ajax': ajax} if pattern_res: # 分段下载时，才每页显示500条 return 500 else: return 50 @list_per_page.setter def list_per_page(self,x): return x model_icon = 'fa fa-commenting' show_all_rel_details = False
4-4.xadmin/views/list.py
@filter_hook def get_context(self): """ Prepare the context for templates. """ self.title = _('%s List') % force_text(self.opts.verbose_name) model_fields = [(f, f.name in self.list_display, self.get_check_field_url(f)) for f in (list(self.opts.fields) + self.get_model_method_fields()) if f.name not in self.list_exclude] new_context = { 'model_name': force_text(self.opts.verbose_name_plural), 'title': self.title, 'cl': self, 'model_fields': model_fields, 'clean_select_field_url': self.get_query_string(remove=[COL_LIST_VAR]), 'has_add_permission': self.has_add_permission(), 'app_label': self.app_label, 'brand_name': self.opts.verbose_name_plural, 'brand_icon': self.get_model_icon(self.model), 'add_url': self.model_admin_url('add'), 'result_headers': self.result_headers(), 'results': self.results(), 'result_count':self.result_count,# 将查询总数携带在context中 } context = super(ListAdminView, self).get_context() context.update(new_context) return context