Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@app.route('/api/files', methods=['GET'])
@app.catch_err
def get_files():
schema = request.args.get('schema', '')
path = request.args.get('path', '')
size = int(request.args.get('size', '0'))
offset = int(request.args.get('offset', '0'))
_, filename = path.split(path)
path = get_obj_path(schema, path)
if not path:
return app.json_error(
HTTPStatus.NOT_FOUND, path=path,
err='illegal path prefix or schema')
try:
@app.route('/api/submit', methods=['POST'])
@app.route('/api/submit/', methods=['POST'])
@app.route('/api/submit_job', methods=['POST'])
@app.route('/api/submit_job/', methods=['POST'])
@app.catch_err
def submit_job():
try:
data: dict = request.get_json(force=True)
except ValueError:
return app.json_error(HTTPStatus.BAD_REQUEST, reason='bad JSON body')
app.logger.info('submit_job: {}'.format(data))
return _submit.submit(data)
@app.route('/api/build/status/', methods=['GET'])
@app.catch_err
def build_status():
name = request.args.get('name', '')
project = request.args.get('project', '')
tag = request.args.get('tag', '')
offset = int(request.args.get('offset', '0'))
logs = strtobool(request.args.get('logs', 'on'))
fn = app.db.get_function(name, project, tag)
if not fn:
return app.json_error(
HTTPStatus.NOT_FOUND, name=name, project=project, tag=tag)
state = get_in(fn, 'status.state', '')
pod = get_in(fn, 'status.build_pod', '')
image = get_in(fn, 'spec.build.image', '')
@app.route('/api/build/function/', methods=['POST'])
@app.catch_err
def build_function():
try:
data = request.get_json(force=True)
except ValueError:
return app.json_error(HTTPStatus.BAD_REQUEST, reason='bad JSON body')
app.logger.info('build_function:\n{}'.format(data))
function = data.get('function')
with_mlrun = strtobool(data.get('with_mlrun', 'on'))
ready = False
try:
fn = new_function(runtime=function)
fn.set_db_connection(app.db)
fn.save(versioned=False)
@app.route('/api/artifact//', methods=['DELETE'])
@app.catch_err
def del_artifact(project, uid):
key = request.args.get('key')
if not key:
return app.json_error(HTTPStatus.BAD_REQUEST, reason='missing data')
tag = request.args.get('tag', '')
app.db.del_artifact(key, tag, project)
return jsonify(ok=True)
@app.route('/api/build/status', methods=['GET'])
@app.route('/api/build/status/', methods=['GET'])
@app.catch_err
def build_status():
name = request.args.get('name', '')
project = request.args.get('project', '')
tag = request.args.get('tag', '')
offset = int(request.args.get('offset', '0'))
logs = strtobool(request.args.get('logs', 'on'))
fn = app.db.get_function(name, project, tag)
if not fn:
return app.json_error(
HTTPStatus.NOT_FOUND, name=name, project=project, tag=tag)
state = get_in(fn, 'status.state', '')
pod = get_in(fn, 'status.build_pod', '')
@app.route('/api/projects//artifact-tags', methods=['GET'])
@app.catch_err
def list_artifact_tags(project):
return jsonify(
ok=True,
project=project,
tags=app.db.list_artifact_tags(project),
)
@app.route('/api/run//', methods=['POST'])
@app.catch_err
def store_run(project, uid):
try:
data = request.get_json(force=True)
except ValueError:
return app.json_error(HTTPStatus.BAD_REQUEST, reason='bad JSON body')
app.logger.debug(data)
iter = int(request.args.get('iter', '0'))
app.db.store_run(data, uid, project, iter=iter)
app.app.logger.info('store run: {}'.format(data))
return jsonify(ok=True)
@app.route('/api/start/function', methods=['POST'])
@app.route('/api/start/function/', methods=['POST'])
@app.catch_err
def start_function():
try:
data = request.get_json(force=True)
except ValueError:
return app.json_error(HTTPStatus.BAD_REQUEST, reason='bad JSON body')
app.logger.info('start_function:\n{}'.format(data))
url = data.get('functionUrl')
if not url:
return app.json_error(
HTTPStatus.BAD_REQUEST,
reason='runtime error: functionUrl not specified',
)
@app.route('/api/submit_pipeline/', methods=['POST'])
@app.catch_err
def submit_pipeline():
namespace = request.args.get('namespace', config.namespace)
experiment_name = request.args.get('experiment', 'Default')
run_name = request.args.get('run', '')
run_name = run_name or \
experiment_name + ' ' + datetime.now().strftime('%Y-%m-%d %H-%M-%S')
arguments = {}
arguments_data = request.headers.get('pipeline-arguments')
if arguments_data:
arguments = ast.literal_eval(arguments_data)
app.logger.info('pipeline arguments {}'.format(arguments_data))
ctype = request.content_type
if '/yaml' in ctype: