Sfoglia il codice sorgente

[spark] Unified exception handling in the API

Romain Rigaux 11 anni fa
parent
commit
671bc65459

+ 31 - 31
apps/spark/src/spark/api.py

@@ -25,7 +25,7 @@ from desktop.lib.exceptions_renderable import PopupException
 from desktop.lib.i18n import force_unicode
 from desktop.models import Document2
 
-from spark.models import get_api, Notebook
+from spark.models import get_api, Notebook, QueryExpired, SessionExpired
 
 
 LOG = logging.getLogger(__name__)
@@ -56,12 +56,12 @@ def execute(request):
   try:
     response['handle'] = get_api(request.user, snippet).execute(notebook, snippet)
     response['status'] = 0
+  except SessionExpired, e:
+    response['status'] = -2    
+  except QueryExpired, e:
+    response['status'] = -3    
   except Exception, e:
-    message = force_unicode(str(e))
-    if 'session not found' in message:
-      response['status'] = -2
-    else:
-      response['message'] = force_unicode(str(e))
+    response['message'] = force_unicode(str(e))
 
   return HttpResponse(json.dumps(response), mimetype="application/json")
 
@@ -75,12 +75,12 @@ def check_status(request):
   try:
     response['query_status'] = get_api(request.user, snippet).check_status(notebook, snippet)
     response['status'] = 0
+  except SessionExpired, e:
+    response['status'] = -2    
+  except QueryExpired, e:
+    response['status'] = -3    
   except Exception, e:
-    message = force_unicode(str(e))
-    if 'session not found' in message: # if 'Invalid OperationHandle' in message --> expired
-      response['status'] = -2
-    else:
-      response['message'] = force_unicode(str(e))
+    response['message'] = force_unicode(str(e))
 
   return HttpResponse(json.dumps(response), mimetype="application/json")
 
@@ -96,12 +96,12 @@ def fetch_result_data(request):
   try:
     response['result'] = get_api(request.user, snippet).fetch_result(notebook, snippet, rows, start_over)
     response['status'] = 0
+  except SessionExpired, e:
+    response['status'] = -2    
+  except QueryExpired, e:
+    response['status'] = -3    
   except Exception, e:
-    message = force_unicode(str(e))
-    if 'session not found' in message:
-      response['status'] = -2
-    else:
-      response['message'] = force_unicode(str(e))
+    response['message'] = force_unicode(str(e))
 
   return HttpResponse(json.dumps(response), mimetype="application/json")
 
@@ -115,12 +115,12 @@ def fetch_result_metadata(request):
   try:
     response['result'] = get_api(request.user, snippet).fetch_result_metadata(notebook, snippet)
     response['status'] = 0
+  except SessionExpired, e:
+    response['status'] = -2    
+  except QueryExpired, e:
+    response['status'] = -3    
   except Exception, e:
-    message = force_unicode(str(e))
-    if 'session not found' in message:
-      response['status'] = -2
-    else:
-      response['message'] = force_unicode(str(e))
+    response['message'] = force_unicode(str(e))
 
   return HttpResponse(json.dumps(response), mimetype="application/json")
 
@@ -134,12 +134,12 @@ def cancel_statement(request):
   try:
     response['result'] = get_api(request.user, snippet).cancel(notebook, snippet)
     response['status'] = 0
+  except SessionExpired, e:
+    response['status'] = -2    
+  except QueryExpired, e:
+    response['status'] = -3    
   except Exception, e:
-    message = force_unicode(str(e))
-    if 'session not found' in message:
-      response['status'] = -2
-    else:
-      response['message'] = force_unicode(str(e))
+    response['message'] = force_unicode(str(e))
 
   return HttpResponse(json.dumps(response), mimetype="application/json")
 
@@ -155,12 +155,12 @@ def get_logs(request):
     response['logs'] = db.get_log(snippet)
     response['progress'] = db._progress(snippet, response['logs']) if snippet['status'] != 'available' else 100
     response['status'] = 0
+  except SessionExpired, e:
+    response['status'] = -2    
+  except QueryExpired, e:
+    response['status'] = -3    
   except Exception, e:
-    message = force_unicode(str(e))
-    if 'session not found' in message:
-      response['status'] = -2
-    else:
-      response['message'] = force_unicode(str(e))
+    response['message'] = force_unicode(str(e))
 
   return HttpResponse(json.dumps(response), mimetype="application/json")
 

+ 58 - 9
apps/spark/src/spark/models.py

@@ -18,6 +18,9 @@
 import json
 import re
 
+from desktop.lib.i18n import smart_str, force_unicode
+from desktop.lib.rest.http_client import RestException
+
 from beeswax import models as beeswax_models
 from beeswax.design import hql_query
 from beeswax.models import QUERY_TYPES, HiveServerQueryHandle, QueryHistory
@@ -26,8 +29,16 @@ from beeswax.server import dbms
 from beeswax.server.dbms import get_query_server_config
 
 from spark.job_server_api import get_api as get_spark_api
-from desktop.lib.i18n import smart_str
-from desktop.lib.rest.http_client import RestException
+
+
+# To move to Editor API
+class SessionExpired(Exception):
+  pass
+
+
+class QueryExpired(Exception):
+  pass
+
 
 
 class Notebook():
@@ -140,7 +151,15 @@ class HS2Api():
     db = self._get_db(snippet)
       
     handle = self._get_handle(snippet)
-    status =  db.get_state(handle)
+
+    try:
+      status =  db.get_state(handle)
+    except Exception, e:
+      message = force_unicode(str(e))
+      if 'Invalid query handle' in message or 'Invalid OperationHandle' in message:
+        raise QueryExpired(e)
+      else:
+        raise e
 
     return {
         'status':
@@ -155,7 +174,15 @@ class HS2Api():
     db = self._get_db(snippet)
       
     handle = self._get_handle(snippet)
-    results = db.fetch(handle, start_over=start_over, rows=rows)
+    
+    try:
+      results = db.fetch(handle, start_over=start_over, rows=rows)
+    except Exception, e:
+      message = force_unicode(str(e))
+      if 'Invalid query handle' in message or 'Invalid OperationHandle' in message:
+        raise QueryExpired(e)
+      else:
+        raise e
     
     # no escaping...
     return {
@@ -195,7 +222,7 @@ class HS2Api():
     elif snippet['type'] == 'impala':
       match = re.search('(\d+)% Complete', logs, re.MULTILINE)
       return int(match.group(1)) if match else 0
-    else: #'spark-sql'
+    else:
       return 50
 
 
@@ -215,20 +242,42 @@ class SparkApi():  # Pig, DBquery, Phoenix...
     api = get_spark_api(self.user)
     session = _get_snippet_session(notebook, snippet)
     
-    return {'id': api.submit_statement(session['id'], snippet['statement']).split('cells/')[1]}
+    try:
+      return {'id': api.submit_statement(session['id'], snippet['statement']).split('cells/')[1]}
+    except Exception, e:
+      message = force_unicode(str(e))
+      if 'session not found' in message:
+        raise SessionExpired(e)
+      else:
+        raise e
 
   def check_status(self, notebook, snippet):
-    return {'status': 'available'}
+    try:
+      return {'status': 'available'}
+    except Exception, e:
+      message = force_unicode(str(e))
+      if 'session not found' in message:
+        raise SessionExpired(e)
+      else:
+        raise e
 
   def fetch_result(self, notebook, snippet, rows, start_over):
     api = get_spark_api(self.user)
     session = _get_snippet_session(notebook, snippet)
     cell = snippet['result']['handle']['id']  
     
-    data = api.fetch_data(session['id'], cell)
+
+    try:
+      data = api.fetch_data(session['id'], cell)
+    except Exception, e:
+      message = force_unicode(str(e))
+      if 'session not found' in message:
+        raise SessionExpired(e)
+      else:
+        raise e
       
     return {
-        'data': [data['output']],
+        'data': [data['output']] if start_over else [], # start_over not supported
         'meta': [{'name': 'Header', 'type': 'String', 'comment': ''}]
     }
 

+ 2 - 0
apps/spark/static/js/spark.vm.js

@@ -266,6 +266,8 @@ var Snippet = function (notebook, snippet) {
         }
       } else if (data.status == -2) {
         self.create_session();
+      } else if (data.status == -3) {
+        self.status('expired');
       } else {
         $(document).trigger("error", data.message);
       }