Merge pull request #1038 from tandreas/csv_export

Implement CSV export for BaseModelView.

doc/introduction.rst

@@ -298,6 +298,12 @@ To **manage related models inline**::
 These inline forms can be customised. Have a look at the API documentation for
 :meth:`~flask_admin.contrib.sqla.ModelView.inline_models`.
 
+To **enable csv export** of the model view::
+
+    can_export = True
+
+This will add a button to the model view that exports records, truncating at :attr:`~flask_admin.model.BaseModelView.max_export_rows`.
+
 Adding Your Own Views
 =====================
 

flask_admin/_compat.py

@@ -31,6 +31,10 @@ if not PY2:
 
         return str(s)
 
+    def csv_encode(s):
+        ''' Returns unicode string expected by Python 3's csv module '''
+        return as_unicode(s)
+
     # Various tools
     from functools import reduce
     from urllib.parse import urljoin, urlparse
@@ -50,6 +54,10 @@ else:
 
         return unicode(s)
 
+    def csv_encode(s):
+        ''' Returns byte string expected by Python 2's csv module '''
+        return as_unicode(s).encode('utf-8')
+
     # Helpers
     reduce = __builtins__['reduce'] if isinstance(__builtins__, dict) else __builtins__.reduce
     from urlparse import urljoin, urlparse

flask_admin/model/typefmt.py

@@ -49,3 +49,8 @@ BASE_FORMATTERS = {
     bool: bool_formatter,
     list: list_formatter,
 }
+
+EXPORT_FORMATTERS = {
+    type(None): empty_formatter,
+    list: list_formatter,
+}

flask_admin/templates/bootstrap2/admin/model/list.html

@@ -26,6 +26,12 @@
         </li>
         {% endif %}
 
+        {% if admin_view.can_export %}
+        <li>
+            <a href="{{ get_url('.export_csv', **request.args) }}" title="{{ _gettext('Export') }}">{{ _gettext('Export') }}</a>
+        </li>
+        {% endif %}
+
         {% if filters %}
         <li class="dropdown">
             {{ model_layout.filter_options() }}

flask_admin/templates/bootstrap3/admin/model/list.html

@@ -26,6 +26,12 @@
         </li>
         {% endif %}
 
+        {% if admin_view.can_export %}
+        <li>
+            <a href="{{ get_url('.export_csv', **request.args) }}" title="{{ _gettext('Export') }}">{{ _gettext('Export') }}</a>
+        </li>
+        {% endif %}
+
         {% if filters %}
         <li class="dropdown">
             {{ model_layout.filter_options() }}

flask_admin/tests/test_model.py

@@ -12,6 +12,8 @@ from wtforms import fields
 from flask_admin import Admin, form
 from flask_admin._compat import iteritems, itervalues
 from flask_admin.model import base, filters
+from flask_admin.model.template import macro
+from itertools import islice
 
 
 def wtforms2_and_up(func):
@@ -46,8 +48,8 @@ class SimpleFilter(filters.BaseFilter):
 
 
 class MockModelView(base.BaseModelView):
-    def __init__(self, model, name=None, category=None, endpoint=None, url=None,
-                 **kwargs):
+    def __init__(self, model, data=None, name=None, category=None,
+                 endpoint=None, url=None, **kwargs):
         # Allow to set any attributes from parameters
         for k, v in iteritems(kwargs):
             setattr(self, k, v)
@@ -60,9 +62,12 @@ class MockModelView(base.BaseModelView):
 
         self.search_arguments = []
 
-        self.all_models = {1: Model(1),
-                           2: Model(2)}
-        self.last_id = 3
+        if data is None:
+            self.all_models = {1: Model(1), 2: Model(2)}
+        else:
+            self.all_models = data
+
+        self.last_id = len(self.all_models) + 1
 
     # Scaffolding
     def get_pk_value(self, model):
@@ -89,9 +94,12 @@ class MockModelView(base.BaseModelView):
         return Form
 
     # Data
-    def get_list(self, page, sort_field, sort_desc, search, filters):
+    def get_list(self, page, sort_field, sort_desc, search, filters,
+                 page_size=None):
         self.search_arguments.append((page, sort_field, sort_desc, search, filters))
-        return len(self.all_models), itervalues(self.all_models)
+        count = len(self.all_models)
+        data = islice(itervalues(self.all_models), 0, page_size)
+        return count, data
 
     def get_one(self, id):
         return self.all_models.get(int(id))
@@ -538,3 +546,120 @@ def check_class_name():
 
     view = DummyView(Model)
     eq_(view.name, 'Dummy View')
+
+
+def test_export_csv():
+    app, admin = setup()
+    client = app.test_client()
+
+    # test redirect when csv export is disabled
+    view = MockModelView(Model, column_list=['col1', 'col2'], endpoint="test")
+    admin.add_view(view)
+
+    rv = client.get('/admin/test/export/csv/')
+    eq_(rv.status_code, 302)
+
+    # basic test of csv export with a few records
+    view_data = {
+        1: Model(1, "col1_1", "col2_1"),
+        2: Model(2, "col1_2", "col2_2"),
+        3: Model(3, "col1_3", "col2_3"),
+    }
+
+    view = MockModelView(Model, view_data, can_export=True,
+                         column_list=['col1', 'col2'])
+    admin.add_view(view)
+
+    rv = client.get('/admin/model/export/csv/')
+    data = rv.data.decode('utf-8')
+    eq_(rv.mimetype, 'text/csv')
+    eq_(rv.status_code, 200)
+    ok_("Col1,Col2\r\n"
+        "col1_1,col2_1\r\n"
+        "col1_2,col2_2\r\n"
+        "col1_3,col2_3\r\n" == data)
+
+    # test utf8 characters in csv export
+    view_data[4] = Model(1, u'\u2013ut8_1\u2013', u'\u2013utf8_2\u2013')
+    view = MockModelView(Model, view_data, can_export=True,
+                         column_list=['col1', 'col2'], endpoint="utf8")
+    admin.add_view(view)
+
+    rv = client.get('/admin/utf8/export/csv/')
+    data = rv.data.decode('utf-8')
+    eq_(rv.status_code, 200)
+    ok_(u'\u2013ut8_1\u2013,\u2013utf8_2\u2013\r\n' in data)
+
+    # test row limit
+    view_data = {
+        1: Model(1, "col1_1", "col2_1"),
+        2: Model(2, "col1_2", "col2_2"),
+        3: Model(3, "col1_3", "col2_3"),
+    }
+
+    view = MockModelView(Model, view_data, can_export=True,
+                         column_list=['col1', 'col2'], export_max_rows=2,
+                         endpoint='row_limit_2')
+    admin.add_view(view)
+
+    rv = client.get('/admin/row_limit_2/export/csv/')
+    data = rv.data.decode('utf-8')
+    eq_(rv.status_code, 200)
+    ok_("Col1,Col2\r\n"
+        "col1_1,col2_1\r\n"
+        "col1_2,col2_2\r\n" == data)
+
+    # test None type, integer type, column_labels, and column_formatters
+    view_data = {
+        1: Model(1, "col1_1", 1),
+        2: Model(2, "col1_2", 2),
+        3: Model(3, None, 3),
+    }
+
+    view = MockModelView(
+        Model, view_data, can_export=True, column_list=['col1', 'col2'],
+        column_labels={'col1': 'Str Field', 'col2': 'Int Field'},
+        column_formatters=dict(col2=lambda v, c, m, p: m.col2*2),
+        endpoint="types_and_formatters"
+    )
+    admin.add_view(view)
+
+    rv = client.get('/admin/types_and_formatters/export/csv/')
+    data = rv.data.decode('utf-8')
+    eq_(rv.status_code, 200)
+    ok_("Str Field,Int Field\r\n"
+        "col1_1,2\r\n"
+        "col1_2,4\r\n"
+        ",6\r\n" == data)
+
+    # test column_formatters_export and column_formatters_export
+    type_formatters = {type(None): lambda view, value: "null"}
+
+    view = MockModelView(
+        Model, view_data, can_export=True, column_list=['col1', 'col2'],
+        column_formatters_export=dict(col2=lambda v, c, m, p: m.col2*3),
+        column_formatters=dict(col2=lambda v, c, m, p: m.col2*2),  # overridden
+        column_type_formatters_export=type_formatters,
+        endpoint="export_types_and_formatters"
+    )
+    admin.add_view(view)
+
+    rv = client.get('/admin/export_types_and_formatters/export/csv/')
+    data = rv.data.decode('utf-8')
+    eq_(rv.status_code, 200)
+    ok_("Col1,Col2\r\n"
+        "col1_1,3\r\n"
+        "col1_2,6\r\n"
+        "null,9\r\n" == data)
+
+    # Macros are not implemented for csv export yet and will throw an error
+    view = MockModelView(
+        Model, can_export=True, column_list=['col1', 'col2'],
+        column_formatters=dict(col1=macro('render_macro')),
+        endpoint="macro_exception"
+    )
+    admin.add_view(view)
+
+    rv = client.get('/admin/macro_exception/export/csv/')
+    data = rv.data.decode('utf-8')
+    eq_(rv.status_code, 500)