replace sqlalchemy core api with sqlhelp

rework_ui/blueprint.py

@@ -14,16 +14,14 @@ from flask import (
     url_for
 )
 
-from sqlalchemy import select
-
 from pml import HTML
 from pygments import highlight
 from pygments.lexers import PythonTracebackLexer
 from pygments.formatters import HtmlFormatter
 
+from sqlhelp import select, update
 from rework import api
 from rework.helper import utcnow
-from rework.schema import task, worker, operation, monitor
 from rework.task import Task
 
 from rework_ui.helper import argsdict
@@ -114,17 +112,21 @@ def reworkui(engine,
         if t is None:
             return json.dumps(0)
 
-        servicename, hostid, domain = engine.execute(
-            select([operation.c.name, operation.c.host, operation.c.domain]
-            ).where(operation.c.id == task.c.operation
-            ).where(task.c.id == t.tid)
-        ).fetchone()
+        op = select(
+            'name', 'host', 'domain'
+        ).table(
+            'rework.operation'
+        ).join(
+            'rework.task as task on (task.operation = operation.id)'
+        ).where(
+            'task.id = %(tid)s', tid=t.tid
+        ).do(engine).fetchone()
 
         newtask = api.schedule(engine,
-                               servicename,
+                               op.name,
                                rawinputdata=t.raw_input,
-                               domain=domain,
-                               hostid=hostid,
+                               domain=op.domain,
+                               hostid=op.host,
                                metadata=t.metadata)
         return json.dumps(newtask.tid)
 
@@ -199,15 +201,17 @@ def reworkui(engine,
     @bp.route('/shutdown-worker/<wid>')
     def shutdown_worker(wid):
         with engine.begin() as cn:
-            cn.execute(worker.update().where(worker.c.id == wid
-            ).values(shutdown=True))
+            update('rework.worker').where(id=wid).values(
+                shutdown=True
+            ).do(cn)
         return json.dumps(True)
 
     @bp.route('/kill-worker/<wid>')
     def kill_worker(wid):
         with engine.begin() as cn:
-            cn.execute(worker.update().where(worker.c.id == wid
-            ).values(kill=True))
+            update('rework.worker').where(id=wid).values(
+                kill=True
+            ).do(cn)
         return json.dumps(True)
 
     class uiargsdict(argsdict):
@@ -218,23 +222,30 @@ def reworkui(engine,
     @bp.route('/workers-table')
     def list_workers():
         # workers
-        sql = select([worker.c.id, worker.c.host, worker.c.domain, worker.c.pid,
-                      worker.c.mem, worker.c.shutdown, worker.c.kill,
-                      worker.c.debugport, worker.c.started]
-        ).order_by(worker.c.id
-        ).where(worker.c.running == True)
+        q = select(
+            'id', 'host', 'domain', 'pid', 'mem', 'shutdown',
+            'kill', 'debugport', 'started'
+        ).table('rework.worker'
+        ).where('running = true'
+        ).order('id')
+
         domain = uiargsdict(request.args).domain
         if domain != 'all':
-            sql = sql.where(worker.c.domain == domain)
+            q.where(domain=domain)
 
-        workers = engine.execute(sql).fetchall()
+        workers = q.do(engine).fetchall()
 
         # monitors
-        sql = select([monitor.c.id, monitor.c.domain, monitor.c.lastseen, monitor.c.options])
+        q = select(
+            'id', 'domain', 'lastseen', 'options'
+        ).table('rework.monitor')
         if domain != 'all':
-            sql = sql.where(monitor.c.domain == domain)
+            q.where(domain=domain)
 
-        monitors = {row.domain: row for row in engine.execute(sql).fetchall()}
+        monitors = {
+            row.domain: row
+            for row in q.do(engine).fetchall()
+        }
         now = utcnow().astimezone(TZ)
 
         h = HTML()
@@ -364,13 +375,14 @@ def reworkui(engine,
     @bp.route('/services-table')
     def list_services():
         args = uiargsdict(request.args)
-        sql = select([operation.c.id, operation.c.host, operation.c.name,
-                      operation.c.path, operation.c.domain]
-        ).order_by(operation.c.domain, operation.c.name)
+        q = select(
+            'id', 'host', 'name', 'path', 'domain'
+        ).table('rework.operation'
+        ).order('domain, name')
         if args.domain != 'all':
-            sql = sql.where(operation.c.domain == args.domain)
+            q.where(domain=args.domain)
 
-        ops = engine.execute(sql)
+        ops = q.do(engine)
         h = HTML()
         h.br()
         with h.table(klass='table table-sm table-bordered table-striped table-hover') as t:

rework_ui/schema.py

-from sqlalchemy import Table, Column, Integer, String
+from pathlib import Path
 
-from rework.schema import meta
+from sqlhelp import sqlfile
 
 
-taskstable = Table(
-    'taskstable', meta,
-    Column('id', Integer, primary_key=True),
-    Column('domain', String, default='default', index=True),
-    Column('hash', String, nullable=False, index=True),
-    Column('content', String, nullable=False),
-    schema='rework'
-)
+SCHEMAFILE = Path(__file__).parent / 'schema.sql'
 
 
 def init(engine):
+    sql = sqlfile(SCHEMAFILE, ns='rework')
     with engine.begin() as cn:
-        taskstable.create(cn)
-
-
-def reset(engine):
-    with engine.begin() as cn:
-        taskstable.drop(cn, checkfirst=True)
+        cn.execute(sql)

rework_ui/schema.sql

+create table {ns}.taskstable (
+  id serial primary key,
+  domain text default 'default',
+  hash text not null,
+  content text not null
+);
+
+create index ix_{ns}_taskstable_domain on {ns}.taskstable (domain);
+create index ix_{ns}_taskstable_hash on {ns}.taskstable (hash);

rework_ui/taskstable.py

@@ -4,25 +4,18 @@ from time import sleep
 from pkg_resources import iter_entry_points
 
 from pml import HTML
-from sqlalchemy import select, desc
-
+from sqlhelp import select, insert
 from rework.task import Task
-from rework.schema import task, operation
-
-from rework_ui.schema import taskstable
 
 
 def latest_table_hash(engine, domain):
-    sql = select(
-        [taskstable.c.hash]
-    ).order_by(
-        desc(taskstable.c.id)
-    ).limit(
-        1
-    ).where(
-        taskstable.c.domain == domain
-    )
-    return engine.execute(sql).scalar()
+    q = select(
+        'hash'
+    ).table('rework.taskstable'
+    ).order('hash'
+    ).limit(1
+    ).where(domain=domain)
+    return q.do(engine).scalar()
 
 
 def refresh_tasks(engine, inithash, domain):
@@ -30,22 +23,20 @@ def refresh_tasks(engine, inithash, domain):
     thash = md5(str(taskstates).encode('ascii')).hexdigest()
     if thash != inithash:
         htmltable = generate_tasks_table(engine, taskstates)
-        sql = taskstable.insert().values(
+        q = insert('rework.taskstable').values(
             hash=thash,
             domain=domain,
             content=htmltable
         )
         with engine.begin() as cn:
-            cn.execute(sql)
+            q.do(cn)
         inithash = thash
         # cleanup old tables
-        sql = taskstable.delete().where(
-            taskstable.c.hash != thash
-        ).where(
-            taskstable.c.domain == domain
-        )
+        sql = ('delete from rework.taskstable '
+               'where hash != %(hash)s '
+               'and   domain = %(domain)s')
         with engine.begin() as cn:
-            cn.execute(sql)
+            cn.execute(sql, hash=thash, domain=domain)
     return inithash
 
 
@@ -76,15 +67,14 @@ def refresh_tasks_file(engine, loop=False, sleeptime=2):
 
 def tasks_info(engine, domain):
     with engine.begin() as cn:
-        sql = select(
-            [task.c.id, task.c.status, operation.c.domain]
-        ).order_by(desc(task.c.id)
-        ).where(task.c.operation == operation.c.id)
+        q = select(
+            't.id', 't.status', 'op.domain'
+        ).table('rework.task as t'
+        ).join('rework.operation as op on (op.id = t.operation)'
+        ).order('t.id', 'desc')
         if domain != 'all':
-            sql = sql.where(
-                operation.c.domain == domain
-            )
-        return cn.execute(sql).fetchall()
+            q.where(domain=domain)
+        return q.do(cn).fetchall()
 
 
 MORE_TASKS_ACTIONS = set()

tests/conftest.py

@@ -19,8 +19,7 @@ def engine(request):
     db.setup_local_pg_cluster(request, DATADIR, PORT)
     uri = 'postgresql://localhost:{}/postgres'.format(PORT)
     e = create_engine(uri)
-    reworkschema.reset(e)
-    reworkschema.init(e)
+    reworkschema.init(e, drop=True)
     ruischema.init(e)
     api.freeze_operations(e)
     return e

tests/test_rui.py

@@ -92,6 +92,19 @@ def test_abort(engine, client):
         assert t.aborted
 
 
+def test_relaunch(engine, client):
+    with workers(engine) as mon:
+        res = client.put('/schedule-task/good_job?user=Babar',
+                         upload_files=[('input_file', 'input.xml', b'the file', 'text/xml')])
+        tid = int(res.body)
+        t = Task.byid(engine, tid)
+        t.join()
+        res = client.put(f'/relaunch-task/{tid}')
+        newtid = int(res.body)
+        t2 = Task.byid(engine, newtid)
+        t2.join()
+
+
 def test_task_life_cycle(engine, client, refresh):
     with workers(engine):
         tasks = []
@@ -243,7 +256,7 @@ def test_tasks_table(engine, client, refresh):
         t.join()
         taskstable.refresh_tasks_file(engine)
         res = client.get('/tasks-table-hash?domain=uranus')
-        assert res.text == 'cbcf36e551ad8fdc0aef16fbefd7c6be'
+        assert res.text == '05265be5adad9bb8b0ee50f837535cfa'
         res = client.get('/tasks-table?domain=uranus')
         refpath = DATADIR / 'tasks-table-uranus.html'
         if refresh: