Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def model(self):
om = get_omega(self.args)
name = self.args.get('')
action = self.args.get('')
async = self.args.get('--async')
kwargs_lst = self.args.get('--param')
output = self.args.get('--result')
X = self._ensure_valid_XY(self.args.get(''))
Y = self._ensure_valid_XY(self.args.get(''))
# parse the list of kw=value values
# e.g. key1=val1 key2=val2 => kwargs_lst = ['key1=val1', 'key2=val2']
# => kw_dct = { 'key1': eval('val1'), 'key2': eval('val2') }
kv_dct = {}
for kv in kwargs_lst:
k, v = kv.split('=', 1)
kv_dct[k] = eval(v)
kwargs = {}
if action in ('predict', 'predict_proba',
def status(self):
om = get_omega(self.args)
name = self.args.get('')
meta = om.jobs.metadata(name)
attrs = meta.attributes
runs = attrs.get('job_runs', [])
run_at, triggers = om.jobs.get_schedule(name, only_pending=True)
self.logger.info("Runs:")
for run in runs:
self.logger.info(" {ts} {status} ".format(**run))
self.logger.info("Next scheduled runs:")
for trigger in triggers:
trigger['ts'] = trigger.get('ts', '')
self.logger.info(" {ts} {status} {event-kind} {event}".format(**trigger))
def plugins(self):
om = get_omega(self.args)
for kind, plugincls in om.defaults.OMEGA_STORE_BACKENDS.items():
self.logger.info(kind, plugincls.__doc__)
def plugins(self):
om = get_omega(self.args)
for kind, plugincls in om.defaults.OMEGA_STORE_BACKENDS.items():
self.logger.info(kind, plugincls.__doc__)
def put(self):
om = get_omega(self.args)
script_path = self.args.get('
def get(self):
om = get_omega(self.args)
local = self.args['
def metadata(self):
om = get_omega(self.args)
name = self.args.get('')
self.logger.info(om.jobs.metadata(name).to_json())
def list(self):
om = get_omega(self.args)
raw = self.args.get('--raw', False)
self.logger.info(om.scripts.list(raw=raw))
def result(self):
from celery.result import AsyncResult
om = get_omega(self.args)
task_id = self.args.get('')
result = AsyncResult(task_id, app=om.runtime.celeryapp).get()
self.logger.info(result)