Error handling for ?_sort and ?_sort_desc

Verifies that they match an existing column, and only one or the other option
is provided - refs #189

Eses a new DatasetteError exception that closes #193

datasette/app.py

@@ -45,6 +45,10 @@
 connections = threading.local()
 
 
+class DatasetteError(Exception):
+    pass
+
+
 class RenderMixin(HTTPMethodView):
     def render(self, templates, **context):
         template = self.jinja_env.select_template(templates)
@@ -205,7 +209,7 @@ async def view_get(self, request, name, hash, **kwargs):
                 return response_or_template_contexts
             else:
                 data, extra_template_data, templates = response_or_template_contexts
-        except (sqlite3.OperationalError, InvalidSql) as e:
+        except (sqlite3.OperationalError, InvalidSql, DatasetteError) as e:
             data = {
                 'ok': False,
                 'error': str(e),
@@ -613,12 +617,26 @@ async def data(self, request, name, hash, table):
             search_description = 'search matches "{}"'.format(search)
             params['search'] = search
 
+        info = self.ds.inspect()
+        table_rows = None
+        sortable_columns = set()
+        if not is_view:
+            table_info = info[name]['tables'][table]
+            table_rows = table_info['count']
+            sortable_columns = set(table_info['columns'])
+
         # Allow for custom sort order
         sort = special_args.get('_sort')
         if sort:
+            if sort not in sortable_columns:
+                raise DatasetteError('Cannot sort table by {}'.format(sort))
             order_by = escape_sqlite(sort)
         sort_desc = special_args.get('_sort_desc')
         if sort_desc:
+            if sort_desc not in sortable_columns:
+                raise DatasetteError('Cannot sort table by {}'.format(sort_desc))
+            if sort:
+                raise DatasetteError('Cannot use _sort and _sort_desc at the same time')
             order_by = '{} desc'.format(escape_sqlite(sort_desc))
 
         count_sql = 'select count(*) from {table_name} {where}'.format(
@@ -728,11 +746,6 @@ async def data(self, request, name, hash, table):
         if use_rowid and filter_columns[0] == 'rowid':
             filter_columns = filter_columns[1:]
 
-        info = self.ds.inspect()
-        table_rows = None
-        if not is_view:
-            table_rows = info[name]['tables'][table]['count']
-
         # Pagination next link
         next_value = None
         next_url = None

tests/test_api.py

@@ -400,6 +400,24 @@ def test_sortable_and_filtered(app_client):
     ]
 
 
+def test_sortable_argument_errors(app_client):
+    response = app_client.get(
+        '/test_tables/sortable.json?_sort=badcolumn',
+        gather_request=False
+    )
+    assert 'Cannot sort table by badcolumn' == response.json['error']
+    response = app_client.get(
+        '/test_tables/sortable.json?_sort_desc=badcolumn2',
+        gather_request=False
+    )
+    assert 'Cannot sort table by badcolumn2' == response.json['error']
+    response = app_client.get(
+        '/test_tables/sortable.json?_sort=content&_sort_desc=pk2',
+        gather_request=False
+    )
+    assert 'Cannot use _sort and _sort_desc at the same time' == response.json['error']
+
+
 @pytest.mark.parametrize('path,expected_rows', [
     ('/test_tables/simple_primary_key.json?content=hello', [
         ['1', 'hello'],