[sql lab] allow EXPlAIN queries (#5558)

* [sql lab] allow EXPlAIN queries

closes https://github.com/andialbrecht/sqlparse/issues/421

* typo
This commit is contained in:
Maxime Beauchemin 2018-08-03 15:33:33 -07:00 committed by GitHub
parent faf35b0daa
commit 9331cf79b5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 27 additions and 2 deletions

View File

@ -130,7 +130,7 @@ def execute_sql(
superset_query = SupersetQuery(rendered_query) superset_query = SupersetQuery(rendered_query)
executed_sql = superset_query.stripped() executed_sql = superset_query.stripped()
SQL_MAX_ROWS = app.config.get('SQL_MAX_ROW') SQL_MAX_ROWS = app.config.get('SQL_MAX_ROW')
if not superset_query.is_select() and not database.allow_dml: if not superset_query.is_readonly() and not database.allow_dml:
return handle_error( return handle_error(
'Only `SELECT` statements are allowed against this database') 'Only `SELECT` statements are allowed against this database')
if query.select_as_cta: if query.select_as_cta:

View File

@ -41,6 +41,13 @@ class SupersetQuery(object):
def is_select(self): def is_select(self):
return self._parsed[0].get_type() == 'SELECT' return self._parsed[0].get_type() == 'SELECT'
def is_explain(self):
return self.sql.strip().upper().startswith('EXPLAIN')
def is_readonly(self):
"""Pessimistic readonly, 100% sure statement won't mutate anything"""
return self.is_select() or self.is_explain()
def stripped(self): def stripped(self):
return self.sql.strip(' \t\n;') return self.sql.strip(' \t\n;')

View File

@ -292,9 +292,21 @@ class SupersetTestCase(unittest.TestCase):
""" """
self.assertEquals({'src'}, self.extract_tables(query)) self.assertEquals({'src'}, self.extract_tables(query))
def multistatement(self): def test_multistatement(self):
query = 'SELECT * FROM t1; SELECT * FROM t2' query = 'SELECT * FROM t1; SELECT * FROM t2'
self.assertEquals({'t1', 't2'}, self.extract_tables(query)) self.assertEquals({'t1', 't2'}, self.extract_tables(query))
query = 'SELECT * FROM t1; SELECT * FROM t2;' query = 'SELECT * FROM t1; SELECT * FROM t2;'
self.assertEquals({'t1', 't2'}, self.extract_tables(query)) self.assertEquals({'t1', 't2'}, self.extract_tables(query))
def test_update_not_select(self):
sql = sql_parse.SupersetQuery('UPDATE t1 SET col1 = NULL')
self.assertEquals(False, sql.is_select())
self.assertEquals(False, sql.is_readonly())
def test_explain(self):
sql = sql_parse.SupersetQuery('EXPLAIN SELECT 1')
self.assertEquals(True, sql.is_explain())
self.assertEquals(False, sql.is_select())
self.assertEquals(True, sql.is_readonly())

View File

@ -55,6 +55,12 @@ class SqlLabTests(SupersetTestCase):
data = self.run_sql('SELECT * FROM unexistant_table', '2') data = self.run_sql('SELECT * FROM unexistant_table', '2')
self.assertLess(0, len(data['error'])) self.assertLess(0, len(data['error']))
def test_explain(self):
self.login('admin')
data = self.run_sql('EXPLAIN SELECT * FROM ab_user', '1')
self.assertLess(0, len(data['data']))
def test_sql_json_has_access(self): def test_sql_json_has_access(self):
main_db = self.get_main_database(db.session) main_db = self.get_main_database(db.session)
security_manager.add_permission_view_menu('database_access', main_db.perm) security_manager.add_permission_view_menu('database_access', main_db.perm)