[task] Remove series of extra spaces
@@ -89,6 +89,7 @@ class ExecutionWrapperCallback(object):
self.meta['status'] = status
download_to_file.update_state(task_id=self.uuid, state='PROGRESS', meta=self.meta)
+
# TODO: Add periodic cleanup task
# TODO: UI should be able to close a query that is available, but not expired
@app.task()
@@ -118,11 +119,13 @@ def download_to_file(notebook, snippet, file_format='csv', max_rows=-1, **kwargs
return meta
+
@app.task(ignore_result=True)
def cancel_async(notebook, snippet, **kwargs):
request = _get_request(**kwargs)
get_api(request, snippet).cancel(notebook, snippet)
+
@app.task(ignore_result=True)
def close_statement_async(notebook, snippet, **kwargs):
request = _get_request(**kwargs)
@@ -186,6 +189,7 @@ def download(*args, **kwargs):
return export_csvxls.file_reader(storage.open(_result_key(notebook), 'rb'))
+
# Why we need this:
# 1) There is no way in celery to differentiate between a task that was submitted, but not yet started and a task that has been GCed.
# 2) The client will keep checking for data until the query is expired. The new definition for expired in this case is a task that has been GCed.
@@ -193,6 +197,7 @@ def _patch_status(notebook):
result = download_to_file.AsyncResult(notebook['uuid'])
result.backend.store_result(notebook['uuid'], None, "SUBMITTED")
+
def execute(*args, **kwargs):
notebook = args[0]
snippet = args[1]
@@ -219,6 +224,7 @@ def execute(*args, **kwargs):
)
return resp
+
def check_status(*args, **kwargs):
notebook = args[0]
result = download_to_file.AsyncResult(notebook['uuid'])
@@ -230,6 +236,7 @@ def check_status(*args, **kwargs):
return {'status': STATE_MAP[state]}
+
def get_log(notebook, snippet, startFrom=None, size=None, postdict=None, user_id=None):
result = download_to_file.AsyncResult(notebook['uuid'])
state = result.state
@@ -254,6 +261,7 @@ def get_log(notebook, snippet, startFrom=None, size=None, postdict=None, user_id
output.write(line)
return output.getvalue()
+
def get_jobs(notebook, snippet, logs, **kwargs): # Re implementation to fetch updated guid in download_to_file from DB
result = download_to_file.AsyncResult(notebook['uuid'])
state = result.state
@@ -271,6 +279,7 @@ def get_jobs(notebook, snippet, logs, **kwargs): # Re implementation to fetch up
api = get_api(request, snippet)
return api.get_jobs(notebook, snippet, logs)
+
def progress(notebook, snippet, logs=None, **kwargs):
result = download_to_file.AsyncResult(notebook['uuid'])
state = result.state
@@ -287,6 +296,7 @@ def progress(notebook, snippet, logs=None, **kwargs):
api = get_api(request, snippet)
return api.progress(notebook, snippet, logs=logs)
+
def fetch_result(notebook, snippet, rows, start_over, **kwargs):
result = download_to_file.AsyncResult(notebook['uuid'])
state = result.state
