Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@app.catch_err
def list_schedules():
return jsonify(
ok=True,
schedules=list(app.db.list_schedules())
)
@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.catch_err
def list_runs():
name = request.args.get('name')
uid = request.args.get('uid')
project = request.args.get('project')
labels = request.args.getlist('label')
state = request.args.get('state')
sort = strtobool(request.args.get('sort', 'on'))
iter = strtobool(request.args.get('iter', 'on'))
last = int(request.args.get('last', '0'))
runs = app.db.list_runs(
name=name or None,
uid=uid or None,
project=project or None,
labels=labels,
state=state or None,
@app.catch_err
def get_project(name):
project = app.db.get_project(name)
if not project:
return app.json_error(error=f'project {name!r} not found')
resp = {
'name': project.name,
'description': project.description,
'owner': project.owner,
'source': project.source,
'users': [u.name for u in project.users],
}
return jsonify(ok=True, project=resp)
@app.catch_err
def list_tags(project):
return jsonify(
ok=True,
project=project,
tags=list(app.db.list_tags(project)),
)
@app.catch_err
def add_project():
data = request.get_json(force=True)
for attr in ('name', 'owner'):
if attr not in data:
return app.json_error(error=f'missing {attr!r}')
project_id = app.db.add_project(data)
return jsonify(
ok=True,
id=project_id,
name=data['name'],
)
@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:
ctype = '.yaml'
@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.catch_err
def store_function(project, name):
try:
data = request.get_json(force=True)
except ValueError:
return app.json_error(HTTPStatus.BAD_REQUEST, reason='bad JSON body')
app.logger.debug(data)
tag = request.args.get('tag', '')
app.db.store_function(data, name, project, tag)
return jsonify(ok=True)
@app.catch_err
def list_projects():
full = strtobool(request.args.get('full', 'no'))
fn = db2dict if full else attrgetter('name')
projects = []
for p in app.db.list_projects():
if isinstance(p, str):
projects.append(p)
else:
projects.append(fn(p))
return jsonify(ok=True, projects=projects)