Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@require_POST
@check_document_access_permission()
@api_error_handler
def check_status(request):
response = {'status': -1}
notebook = json.loads(request.POST.get('notebook', '{}'))
snippet = json.loads(request.POST.get('snippet', '{}'))
if not snippet:
nb_doc = Document2.objects.get_by_uuid(user=request.user, uuid=notebook['id'])
notebook = Notebook(document=nb_doc).get_data()
snippet = notebook['snippets'][0]
try:
response['query_status'] = get_api(request, snippet).check_status(notebook, snippet)
response['status'] = 0
except SessionExpired:
@api_error_handler
def create_session(request):
response = {'status': -1}
session = json.loads(request.POST.get('session', '{}'))
properties = session.get('properties', [])
response['session'] = get_api(request, session).create_session(lang=session['type'], properties=properties)
response['status'] = 0
return JsonResponse(response)
@require_POST
@check_document_access_permission()
@api_error_handler
def statement_compatibility(request):
response = {'status': -1, 'message': ''}
notebook = json.loads(request.POST.get('notebook', '{}'))
snippet = json.loads(request.POST.get('snippet', '{}'))
source_platform = request.POST.get('sourcePlatform')
target_platform = request.POST.get('targetPlatform')
api = get_api(request, snippet)
response['query_compatibility'] = api.statement_compatibility(notebook, snippet, source_platform=source_platform, target_platform=target_platform)
response['status'] = 0
return JsonResponse(response)
@api_error_handler
def autocomplete(request, server=None, database=None, table=None, column=None, nested=None):
response = {'status': -1}
# Passed by check_document_access_permission but unused by APIs
notebook = json.loads(request.POST.get('notebook', '{}'))
snippet = json.loads(request.POST.get('snippet', '{}'))
try:
autocomplete_data = get_api(request, snippet).autocomplete(snippet, database, table, column, nested)
response.update(autocomplete_data)
except QueryExpired:
pass
response['status'] = 0
return JsonResponse(response)
@require_POST
@check_document_access_permission()
@api_error_handler
def fetch_result_size(request):
response = {'status': -1}
notebook = json.loads(request.POST.get('notebook', '{}'))
snippet = json.loads(request.POST.get('snippet', '{}'))
response['result'] = get_api(request, snippet).fetch_result_size(notebook, snippet)
response['status'] = 0
return JsonResponse(response)
@api_error_handler
def get_external_statement(request):
response = {'status': -1, 'message': ''}
notebook = json.loads(request.POST.get('notebook', '{}'))
snippet = json.loads(request.POST.get('snippet', '{}'))
if snippet.get('statementType') == 'file':
response['statement'] = _get_statement_from_file(request.user, request.fs, snippet)
elif snippet.get('statementType') == 'document':
notebook = Notebook(Document2.objects.get_by_uuid(user=request.user, uuid=snippet['associatedDocumentUuid'], perm_type='read'))
response['statement'] = notebook.get_str()
response['status'] = 0
return JsonResponse(response)
@api_error_handler
@check_document_modify_permission()
def clear_history(request):
response = {'status': -1}
notebook = json.loads(request.POST.get('notebook', '{}'))
doc_type = request.POST.get('doc_type')
is_notification_manager = request.POST.get('is_notification_manager', 'false') == 'true'
if is_notification_manager:
history = Document2.objects.get_tasks_history(user=request.user)
else:
history = Document2.objects.get_history(doc_type='query-%s' % doc_type, user=request.user)
response['updated'] = history.delete()
response['message'] = _('History cleared !')
response['status'] = 0
@api_error_handler
def cancel_statement(request):
response = {'status': -1}
notebook = json.loads(request.POST.get('notebook', '{}'))
snippet = None
operation_id = request.POST.get('operationId') or notebook['uuid']
snippet = _get_snippet(request.user, notebook, snippet, operation_id)
with opentracing.tracer.start_span('notebook-cancel_statement') as span:
response['result'] = get_api(request, snippet).cancel(notebook, snippet)
span.set_tag('user-id', request.user.username)
span.set_tag(
'query-id',
snippet['result']['handle']['guid'] if snippet['result'].get('handle') and snippet['result']['handle'].get('guid') else None
@api_error_handler
def explain(request):
response = {'status': -1}
notebook = json.loads(request.POST.get('notebook', '{}'))
snippet = json.loads(request.POST.get('snippet', '{}'))
response = get_api(request, snippet).explain(notebook, snippet)
return JsonResponse(response)
@api_error_handler
def statement_similarity(request):
response = {'status': -1, 'message': ''}
notebook = json.loads(request.POST.get('notebook', '{}'))
snippet = json.loads(request.POST.get('snippet', '{}'))
source_platform = request.POST.get('sourcePlatform')
api = get_api(request, snippet)
response['statement_similarity'] = api.statement_similarity(notebook, snippet, source_platform=source_platform)
response['status'] = 0
return JsonResponse(response)