96 lines
3.3 KiB
Python
96 lines
3.3 KiB
Python
"""SQLite-backed durable job queue. Single-worker, multiple API connections."""
|
|
import sqlite3
|
|
import time
|
|
from pathlib import Path
|
|
|
|
DATA = Path('/data')
|
|
DB_PATH = DATA / 'queue.sqlite3'
|
|
|
|
FIELDS = {
|
|
'title', 'source_type', 'source', 'format', 'state', 'progress', 'speed',
|
|
'total', 'completed', 'part_index', 'gid', 'error', 'output', 'note', 'updated',
|
|
}
|
|
|
|
|
|
def connection():
|
|
DATA.mkdir(parents=True, exist_ok=True)
|
|
con = sqlite3.connect(DB_PATH, timeout=30)
|
|
con.row_factory = sqlite3.Row
|
|
con.execute('PRAGMA busy_timeout=30000')
|
|
con.execute('PRAGMA journal_mode=WAL')
|
|
return con
|
|
|
|
|
|
def init_db():
|
|
with connection() as con:
|
|
con.execute('''CREATE TABLE IF NOT EXISTS jobs (
|
|
id TEXT PRIMARY KEY,
|
|
title TEXT NOT NULL,
|
|
source_type TEXT NOT NULL,
|
|
source TEXT NOT NULL,
|
|
format TEXT NOT NULL,
|
|
state TEXT NOT NULL DEFAULT 'queued',
|
|
progress REAL NOT NULL DEFAULT 0,
|
|
speed INTEGER NOT NULL DEFAULT 0,
|
|
total INTEGER NOT NULL DEFAULT 0,
|
|
completed INTEGER NOT NULL DEFAULT 0,
|
|
part_index INTEGER NOT NULL DEFAULT 0,
|
|
gid TEXT NOT NULL DEFAULT '',
|
|
error TEXT NOT NULL DEFAULT '',
|
|
output TEXT NOT NULL DEFAULT '',
|
|
note TEXT NOT NULL DEFAULT '',
|
|
created REAL NOT NULL,
|
|
updated REAL NOT NULL
|
|
)''')
|
|
con.execute('CREATE INDEX IF NOT EXISTS jobs_created_idx ON jobs(created)')
|
|
# Zadania przerwane w trakcie rozpakowania/konwersji moga byc ponowione.
|
|
con.execute("UPDATE jobs SET state='preparing', note='Wznowiono konwersje po restarcie' "
|
|
"WHERE state IN ('extracting','converting')")
|
|
con.execute("UPDATE jobs SET state='queued', note='Przywrocono zadanie po restarcie' "
|
|
"WHERE state='vimm_browser'")
|
|
|
|
|
|
def create(job):
|
|
now = time.time()
|
|
with connection() as con:
|
|
con.execute('''INSERT INTO jobs(id,title,source_type,source,format,state,created,updated)
|
|
VALUES(?,?,?,?,?,'queued',?,?)''',
|
|
(job['id'], job['title'], job['source_type'],
|
|
job['source'], job['format'], now, now))
|
|
|
|
|
|
def get(job_id):
|
|
with connection() as con:
|
|
row = con.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def list_jobs():
|
|
with connection() as con:
|
|
rows = con.execute('SELECT * FROM jobs ORDER BY created DESC LIMIT 300').fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def next_job():
|
|
with connection() as con:
|
|
row = con.execute('''SELECT * FROM jobs
|
|
WHERE state IN ('queued','downloading','preparing')
|
|
ORDER BY created ASC LIMIT 1''').fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def update(job_id, **values):
|
|
if not values:
|
|
return
|
|
if not set(values).issubset(FIELDS):
|
|
raise ValueError('Niepoprawne pola aktualizacji bazy')
|
|
values['updated'] = time.time()
|
|
with connection() as con:
|
|
con.execute('UPDATE jobs SET ' + ','.join(f'{k}=?' for k in values)
|
|
+ ' WHERE id=?', (*values.values(), job_id))
|
|
|
|
|
|
def erase(job_id):
|
|
with connection() as con:
|
|
con.execute('DELETE FROM jobs WHERE id=?', (job_id,))
|