Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def api_set_meta():
if get_user() is None:
return "Authentication required", 401
query = g.conn.session.query(Query).filter(Query.id == request.form['query_id']).one()
if query.user_id != get_user().id:
return "Authorization denied", 403
if 'title' in request.form:
query.title = request.form['title']
if 'published' in request.form:
query.published = request.form['published'] == '1'
if 'description' in request.form:
query.description = request.form['description']
g.conn.session.add(query)
g.conn.session.commit()
return json.dumps({'id': query.id})
def unstar_query():
if get_user() is None:
return "Unauthorized access", 403
query = g.conn.session.query(Query).get(request.form['query_id'])
if query:
star = g.conn.session.query(Star)\
.filter(Star.query_id == request.form['query_id'])\
.filter(Star.user_id == get_user().id)\
.one()
g.conn.session.delete(star)
g.conn.session.commit()
return ""
else:
return "Query not found", 404
def unstar_query():
if get_user() is None:
return "Unauthorized access", 403
query = g.conn.session.query(Query).get(request.form['query_id'])
if query:
star = g.conn.session.query(Star)\
.filter(Star.query_id == request.form['query_id'])\
.filter(Star.user_id == get_user().id)\
.one()
g.conn.session.delete(star)
g.conn.session.commit()
return ""
else:
return "Query not found", 404
def api_run_query():
if get_user() is None:
return "Authentication required", 401
text = request.form['text']
query = g.conn.session.query(Query).filter(Query.id == request.form['query_id']).one()
if query.user_id != get_user().id or \
g.conn.session.query(UserGroup).filter(UserGroup.user_id == get_user().id) \
.filter(UserGroup.group_name == 'blocked').first():
return "Authorization denied", 403
if query.latest_rev and query.latest_rev.latest_run:
result = worker.run_query.AsyncResult(query.latest_rev.latest_run.task_id)
if not result.ready():
result.revoke(terminate=True)
query.latest_rev.latest_run.status = QueryRun.STATUS_SUPERSEDED
g.conn.session.add(query.latest_rev.latest_run)
g.conn.session.commit()
query_rev = QueryRevision(query_id=query.id, text=text)
query.latest_rev = query_rev
# XXX (phuedx, 2014/08/08): This deviates from the pre-existing
def new_query():
if get_user() is None:
return redirect("/login?next=/query/new")
query = Query()
query.user = get_user()
g.conn.session.add(query)
g.conn.session.commit()
return redirect(url_for('query_show', query_id=query.id))
def fork_query(id):
if get_user() is None:
return redirect("/login?next=fork/{id}".format(id=id))
query = Query()
query.user = get_user()
parent_query = g.conn.session.query(Query).filter(Query.id == id).one()
query.title = parent_query.title
query.parent_id = parent_query.id
query.description = parent_query.description
g.conn.session.add(query)
g.conn.session.commit()
query_rev = QueryRevision(query_id=query.id, text=parent_query.latest_rev.text)
query.latest_rev = query_rev
g.conn.session.add(query)
g.conn.session.add(query_rev)
g.conn.session.commit()
return redirect(url_for('query_show', query_id=query.id))
def pref_set(key, value):
if get_user() is None:
return "Authentication required", 401
get_preferences()[key] = (None if value == 'null' else value)
return Response(json.dumps({'key': key, 'success': ''})), 201
def index():
return render_template("landing.html", user=get_user())
def query_show(query_id):
query = g.conn.session.query(Query).filter(Query.id == query_id).one()
can_edit = get_user() is not None and get_user().id == query.user_id
is_starred = False
if get_user():
is_starred = g.conn.session.query(func.count(Star.id))\
.filter(Star.user_id == get_user().id)\
.filter(Star.query_id == query_id).scalar() == 1
jsvars = {
'query_id': query.id,
'can_edit': can_edit,
'is_starred': is_starred,
'published': query.published,
'preferences': get_preferences()
}
if query.latest_rev and query.latest_rev.latest_run_id:
jsvars['qrun_id'] = query.latest_rev.latest_run_id
def pref_get(key):
if get_user() is None:
return "Authentication required", 401
if key in get_preferences():
return Response(json.dumps({'key': key, 'value': get_preferences()[key]}))
else:
return Response(json.dumps({'key': key, 'error': 'novalue'}))