Commit fa55effc authored by Julian Rother's avatar Julian Rother

Extended job api to support different queues

parent fc99fba9
......@@ -283,6 +283,7 @@ CREATE TABLE IF NOT EXISTS `jobs` (
`id` INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
`type` text NOT NULL,
`priority` INTEGER NOT NULL DEFAULT 0,
`queue` text NOT NULL DEFAULT 'default',
`state` text NOT NULL DEFAULT 'ready',
`time_finished` datetime DEFAULT '',
`time_scheduled` datetime DEFAULT '',
......
......@@ -58,7 +58,7 @@ def schedule_transcode(source, fmt_id=None, video=None):
data['lecture_id'] = lecture['id']
data['format_id'] = fmt['id']
data['source_id'] = source['id']
schedule_job('transcode', data)
schedule_job('transcode', data, queue="background")
@job_handler('probe-raw')
def update_lecture_videos(jobid, jobtype, data, state, status):
......
......@@ -46,7 +46,7 @@ def jobs_action(action, jobid=None):
query('UPDATE jobs SET state="ready" WHERE state = "failed" AND (id = ? OR ? IS NULL)', jobid, jobid)
if action == 'copy':
if jobid:
query("INSERT INTO jobs SELECT NULL, type, priority, 'ready', '', '' , ?, '', NULL, data, '{}' FROM jobs where ID=?;", datetime.now(), jobid)
query("INSERT INTO jobs SELECT NULL, type, priority, queue, 'ready', '', '' , ?, '', NULL, data, '{}' FROM jobs where ID=?;", datetime.now(), jobid)
if action == 'delete':
if jobid:
query('UPDATE jobs SET state="deleted" WHERE id = ?', jobid)
......@@ -125,8 +125,9 @@ def jobs_schedule(hostname):
while (not job):
query("BEGIN")
for i in query('SELECT * FROM jobs WHERE state = "ready" ORDER BY priority DESC'):
if i['type'] in hostdata['jobtypes'].split(','):
job=i
if i['type'] in hostdata['jobtypes'] and \
i['queue'] in hostdata['queues']:
job = i
break
if not job:
return 'no jobs', 503
......
......@@ -497,11 +497,11 @@ def job_handler(*types, state='finished'):
def date_json_handler(obj):
return obj.isoformat() if hasattr(obj, 'isoformat') else obj
def schedule_job(jobtype, data=None, priority=0):
def schedule_job(jobtype, data=None, priority=0, queue="default"):
if not data:
data = {}
modify('INSERT INTO jobs (type, priority, data, time_created) VALUES (?, ?, ?, ?)',
jobtype, priority, json.dumps(data, default=date_json_handler), datetime.now())
modify('INSERT INTO jobs (type, priority, queue, data, time_created) VALUES (?, ?, ?, ?, ?)',
jobtype, priority, queue, json.dumps(data, default=date_json_handler), datetime.now())
edit_handlers = {}
def edit_handler(*tables, field=None):
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment