|
@@ -15,31 +15,68 @@
|
|
|
# See the License for the specific language governing permissions and
|
|
# See the License for the specific language governing permissions and
|
|
|
# limitations under the License.
|
|
# limitations under the License.
|
|
|
|
|
|
|
|
|
|
+'''
|
|
|
|
|
+SQL Alchemy offers native connections to databases via dialects https://docs.sqlalchemy.org/en/latest/dialects/.
|
|
|
|
|
+
|
|
|
|
|
+When the dialect of a paricular datavase is installed on the Hue API server, any of its URL connection strings should work.
|
|
|
|
|
+
|
|
|
|
|
+e.g.
|
|
|
|
|
+mysql://root:root@localhost:3306/hue
|
|
|
|
|
+
|
|
|
|
|
+To offer more self service capabilities, parts of the URL can be parameterized.
|
|
|
|
|
+
|
|
|
|
|
+Supported parameters are:
|
|
|
|
|
+
|
|
|
|
|
+* USER
|
|
|
|
|
+* PASSWORD
|
|
|
|
|
+
|
|
|
|
|
+e.g.
|
|
|
|
|
+mysql://${USER}:${PASSWORD}@localhost:3306/hue
|
|
|
|
|
+
|
|
|
|
|
+Parameters are not saved at any time in the Hue database. The are currently not even cached in the Hue process. The clients serves these parameters
|
|
|
|
|
+each time a query is sent.
|
|
|
|
|
+
|
|
|
|
|
+Note: the SQL Alchemy engine could leverage create_session() and cache the engine object (without its credentials) like in the jdbc.py interpreter.
|
|
|
|
|
+Note: this is currently supporting concurrent querying by one users as engine is a new object each time. Could use a thread global SQL Alchemy
|
|
|
|
|
+session at some point.
|
|
|
|
|
+Note: using the task server would not leverage any caching.
|
|
|
|
|
+'''
|
|
|
|
|
+
|
|
|
import datetime
|
|
import datetime
|
|
|
import json
|
|
import json
|
|
|
import logging
|
|
import logging
|
|
|
import uuid
|
|
import uuid
|
|
|
|
|
+import sys
|
|
|
|
|
+
|
|
|
|
|
+from string import Template
|
|
|
|
|
|
|
|
-from desktop.lib import export_csvxls
|
|
|
|
|
-from desktop.lib.i18n import force_unicode
|
|
|
|
|
from django.utils.translation import ugettext as _
|
|
from django.utils.translation import ugettext as _
|
|
|
|
|
+from sqlalchemy import create_engine, inspect
|
|
|
|
|
+from sqlalchemy.exc import OperationalError
|
|
|
|
|
|
|
|
|
|
+from desktop.lib import export_csvxls
|
|
|
|
|
+from desktop.lib.i18n import force_unicode
|
|
|
from beeswax import data_export
|
|
from beeswax import data_export
|
|
|
from librdbms.server import dbms
|
|
from librdbms.server import dbms
|
|
|
|
|
|
|
|
-from notebook.connectors.base import Api, QueryError, QueryExpired, _get_snippet_name
|
|
|
|
|
|
|
+from notebook.connectors.base import Api, QueryError, QueryExpired, _get_snippet_name, AuthenticationRequired
|
|
|
from notebook.models import escape_rows
|
|
from notebook.models import escape_rows
|
|
|
-from sqlalchemy import create_engine, inspect
|
|
|
|
|
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
-LOG = logging.getLogger(__name__)
|
|
|
|
|
CONNECTION_CACHE = {}
|
|
CONNECTION_CACHE = {}
|
|
|
|
|
+LOG = logging.getLogger(__name__)
|
|
|
|
|
+
|
|
|
|
|
|
|
|
def query_error_handler(func):
|
|
def query_error_handler(func):
|
|
|
def decorator(*args, **kwargs):
|
|
def decorator(*args, **kwargs):
|
|
|
try:
|
|
try:
|
|
|
return func(*args, **kwargs)
|
|
return func(*args, **kwargs)
|
|
|
|
|
+ except OperationalError, e:
|
|
|
|
|
+ message = str(e)
|
|
|
|
|
+ if '1045' in message: # 'Access denied' # MySQL
|
|
|
|
|
+ raise AuthenticationRequired(message=message)
|
|
|
|
|
+ else:
|
|
|
|
|
+ raise e
|
|
|
except Exception, e:
|
|
except Exception, e:
|
|
|
message = force_unicode(e)
|
|
message = force_unicode(e)
|
|
|
if 'Invalid query handle' in message or 'Invalid OperationHandle' in message:
|
|
if 'Invalid query handle' in message or 'Invalid OperationHandle' in message:
|
|
@@ -51,13 +88,27 @@ def query_error_handler(func):
|
|
|
|
|
|
|
|
|
|
|
|
|
class SqlAlchemyApi(Api):
|
|
class SqlAlchemyApi(Api):
|
|
|
|
|
+
|
|
|
def __init__(self, user, interpreter=None):
|
|
def __init__(self, user, interpreter=None):
|
|
|
|
|
+ self.user = user
|
|
|
self.options = interpreter['options']
|
|
self.options = interpreter['options']
|
|
|
- self.engine = create_engine(self.options['url'])
|
|
|
|
|
|
|
+ self.engine = None # Currently instantiated by an execute()
|
|
|
|
|
|
|
|
@query_error_handler
|
|
@query_error_handler
|
|
|
def execute(self, notebook, snippet):
|
|
def execute(self, notebook, snippet):
|
|
|
guid = uuid.uuid4().hex
|
|
guid = uuid.uuid4().hex
|
|
|
|
|
+
|
|
|
|
|
+ if '${' in self.options['url']: # URL parameters substitution
|
|
|
|
|
+ vars = {'user': self.user.username}
|
|
|
|
|
+ for _prop in self.options['session']['properties']:
|
|
|
|
|
+ if _prop['name'] == 'user':
|
|
|
|
|
+ vars['USER'] = _prop['value']
|
|
|
|
|
+ if _prop['name'] == 'password':
|
|
|
|
|
+ vars['PASSWORD'] = _prop['value']
|
|
|
|
|
+
|
|
|
|
|
+ raw_url = Template(self.options['url'])
|
|
|
|
|
+ self.engine = create_engine(raw_url.safe_substitute(**vars))
|
|
|
|
|
+
|
|
|
connection = self.engine.connect()
|
|
connection = self.engine.connect()
|
|
|
result = connection.execute(snippet['statement'])
|
|
result = connection.execute(snippet['statement'])
|
|
|
cache = {
|
|
cache = {
|
|
@@ -88,6 +139,7 @@ class SqlAlchemyApi(Api):
|
|
|
def check_status(self, notebook, snippet):
|
|
def check_status(self, notebook, snippet):
|
|
|
guid = snippet['result']['handle']['guid']
|
|
guid = snippet['result']['handle']['guid']
|
|
|
connection = CONNECTION_CACHE.get(guid)
|
|
connection = CONNECTION_CACHE.get(guid)
|
|
|
|
|
+
|
|
|
if connection:
|
|
if connection:
|
|
|
return {'status': 'available'}
|
|
return {'status': 'available'}
|
|
|
else:
|
|
else:
|
|
@@ -97,6 +149,7 @@ class SqlAlchemyApi(Api):
|
|
|
def fetch_result(self, notebook, snippet, rows, start_over):
|
|
def fetch_result(self, notebook, snippet, rows, start_over):
|
|
|
guid = snippet['result']['handle']['guid']
|
|
guid = snippet['result']['handle']['guid']
|
|
|
cache = CONNECTION_CACHE.get(guid)
|
|
cache = CONNECTION_CACHE.get(guid)
|
|
|
|
|
+
|
|
|
if cache:
|
|
if cache:
|
|
|
data = cache['result'].fetchmany(rows)
|
|
data = cache['result'].fetchmany(rows)
|
|
|
meta = cache['meta']
|
|
meta = cache['meta']
|
|
@@ -104,6 +157,7 @@ class SqlAlchemyApi(Api):
|
|
|
else:
|
|
else:
|
|
|
data = []
|
|
data = []
|
|
|
meta = []
|
|
meta = []
|
|
|
|
|
+
|
|
|
return {
|
|
return {
|
|
|
'has_more': data and len(data) >= rows,
|
|
'has_more': data and len(data) >= rows,
|
|
|
'data': data if data else [],
|
|
'data': data if data else [],
|
|
@@ -156,24 +210,29 @@ class SqlAlchemyApi(Api):
|
|
|
def download(self, notebook, snippet, format, user_agent=None):
|
|
def download(self, notebook, snippet, format, user_agent=None):
|
|
|
file_name = _get_snippet_name(notebook)
|
|
file_name = _get_snippet_name(notebook)
|
|
|
guid = uuid.uuid4().hex
|
|
guid = uuid.uuid4().hex
|
|
|
|
|
+
|
|
|
connection = self.engine.connect()
|
|
connection = self.engine.connect()
|
|
|
result = connection.execute(snippet['statement'])
|
|
result = connection.execute(snippet['statement'])
|
|
|
|
|
+
|
|
|
CONNECTION_CACHE[guid] = {
|
|
CONNECTION_CACHE[guid] = {
|
|
|
'connection': connection,
|
|
'connection': connection,
|
|
|
'result': result
|
|
'result': result
|
|
|
}
|
|
}
|
|
|
db = FixedResult([col[0] if type(col) is dict or type(col) is tuple else col for col in result.cursor.description])
|
|
db = FixedResult([col[0] if type(col) is dict or type(col) is tuple else col for col in result.cursor.description])
|
|
|
|
|
+
|
|
|
def callback():
|
|
def callback():
|
|
|
connection = CONNECTION_CACHE.get(guid)
|
|
connection = CONNECTION_CACHE.get(guid)
|
|
|
if connection:
|
|
if connection:
|
|
|
connection['connection'].close()
|
|
connection['connection'].close()
|
|
|
del CONNECTION_CACHE[guid]
|
|
del CONNECTION_CACHE[guid]
|
|
|
|
|
+
|
|
|
return data_export.download({'guid': guid}, format, db, id=snippet['id'], file_name=file_name, callback=callback)
|
|
return data_export.download({'guid': guid}, format, db, id=snippet['id'], file_name=file_name, callback=callback)
|
|
|
|
|
|
|
|
|
|
|
|
|
@query_error_handler
|
|
@query_error_handler
|
|
|
def close_statement(self, snippet):
|
|
def close_statement(self, snippet):
|
|
|
result = {'status': -1}
|
|
result = {'status': -1}
|
|
|
|
|
+
|
|
|
try:
|
|
try:
|
|
|
guid = snippet['result']['handle']['guid']
|
|
guid = snippet['result']['handle']['guid']
|
|
|
connection = CONNECTION_CACHE.get('guid')
|
|
connection = CONNECTION_CACHE.get('guid')
|
|
@@ -303,4 +362,4 @@ class FixedResult():
|
|
|
data = connection['result'].fetchmany(rows)
|
|
data = connection['result'].fetchmany(rows)
|
|
|
return FixedResultSet(self.metadata, data, data is not None and len(data) >= rows)
|
|
return FixedResultSet(self.metadata, data, data is not None and len(data) >= rows)
|
|
|
else:
|
|
else:
|
|
|
- return FixedResultSet([], [])
|
|
|
|
|
|
|
+ return FixedResultSet([], [])
|