Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@check_document_access_permission
def copy(request):
response = {'status': -1}
notebooks = json.loads(request.POST.get('notebooks', '[]'))
if len(notebooks) == 0:
response['message'] = _('No notebooks have been selected for copying.')
else:
ctr = 0
failures = []
for notebook in notebooks:
try:
doc2 = Document2.objects.get_by_uuid(user=request.user, uuid=notebook['uuid'])
doc = doc2._get_doc1()
name = doc2.name + '-copy'
doc2 = doc2.copy(name=name, owner=request.user)
@check_document_access_permission
def editor(request, is_mobile=False, is_embeddable=False):
editor_id = request.GET.get('editor')
editor_type = request.GET.get('type', 'hive')
gist_id = request.GET.get('gist')
if editor_type == 'notebook' or request.GET.get('notebook'):
return notebook(request)
if editor_type == 'gist':
gist_doc = _get_gist_document(uuid=gist_id)
editor_type = gist_doc.extra
if editor_id and not gist_id: # Open existing saved editor document
editor_type = _get_editor_type(editor_id)
template = 'editor.mako'
@check_document_access_permission
@api_error_handler
def get_logs(request):
response = {'status': -1}
operation_id = request.POST.get('operationId')
notebook = json.loads(request.POST.get('notebook', '{}'))
snippet = json.loads(request.POST.get('snippet', '{}'))
startFrom = request.POST.get('from')
startFrom = int(startFrom) if startFrom else None
size = request.POST.get('size')
size = int(size) if size else None
full_log = smart_str(request.POST.get('full_log', ''))
snippet = _get_snippet(request.user, notebook, snippet, operation_id)
@check_document_access_permission
@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)
@check_document_access_permission
@api_error_handler
def statement_risk(request):
response = {'status': -1, 'message': ''}
notebook = json.loads(request.POST.get('notebook', '{}'))
snippet = json.loads(request.POST.get('snippet', '{}'))
api = HS2Api(request.user, snippet)
response['query_complexity'] = api.statement_risk(notebook, snippet)
response['status'] = 0
return JsonResponse(response)
@check_document_access_permission
def download(request):
if not ENABLE_DOWNLOAD.get():
return serve_403_error(request)
notebook = json.loads(request.POST.get('notebook', '{}'))
snippet = json.loads(request.POST.get('snippet', '{}'))
file_format = request.POST.get('format', 'csv')
user_agent = request.META.get('HTTP_USER_AGENT')
file_name = _get_snippet_name(notebook)
content_generator = get_api(request, snippet).download(notebook, snippet, file_format=file_format)
response = export_csvxls.make_response(content_generator, file_format, file_name, user_agent=user_agent)
if snippet['id']:
response.set_cookie(
'download-%s' % snippet['id'],
@check_document_access_permission
def close_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)
try:
with opentracing.tracer.start_span('notebook-close_statement') as span:
response['result'] = get_api(request, snippet).close_statement(notebook, snippet)
span.set_tag('user-id', request.user.username)
span.set_tag(
'query-id',
@check_document_access_permission
@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)
@check_document_access_permission
@api_error_handler
def get_sample_data(request, server=None, database=None, table=None, column=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', '{}'))
is_async = json.loads(request.POST.get('async', 'false'))
operation = json.loads(request.POST.get('operation', '"default"'))
sample_data = get_api(request, snippet).get_sample_data(snippet, database, table, column, is_async=is_async, operation=operation)
response.update(sample_data)
response['status'] = 0
return JsonResponse(response)
@check_document_access_permission
def open_notebook(request):
response = {'status': -1}
notebook_id = request.GET.get('notebook')
notebook = Notebook(document=Document2.objects.get(id=notebook_id))
notebook = upgrade_session_properties(request, notebook)
response['status'] = 0
response['notebook'] = notebook.get_json()
response['message'] = _('Notebook loaded successfully')