use transactions

rework_ui/blueprint.py

@@ -189,7 +189,7 @@ def reworkui(engine,
 
     @bp.route('/list_jobs')
     def list_jobs():
-        with engine.connect() as cn:
+        with engine.begin() as cn:
             tsql = 'select id from rework.task order by id'
             jobids = cn.execute(tsql).fetchall()
             opsql = 'select id, name from rework.operation'
@@ -210,14 +210,14 @@ def reworkui(engine,
 
     @bp.route('/shutdown-worker/<wid>')
     def shutdown_worker(wid):
-        with engine.connect() as cn:
+        with engine.begin() as cn:
             cn.execute(worker.update().where(worker.c.id == wid
             ).values(shutdown=True))
         return json.dumps(True)
 
     @bp.route('/kill-worker/<wid>')
     def kill_worker(wid):
-        with engine.connect() as cn:
+        with engine.begin() as cn:
             cn.execute(worker.update().where(worker.c.id == wid
             ).values(kill=True))
         return json.dumps(True)
@@ -308,14 +308,14 @@ def reworkui(engine,
 
     @bp.route('/delete-task/<tid>')
     def delete_task(tid):
-        with engine.connect() as cn:
+        with engine.begin() as cn:
             cn.execute("delete from rework.task where id = %(tid)s and status != 'running'",
                        tid=tid)
         return json.dumps(True)
 
     @bp.route('/abort-task/<tid>')
     def abort_task(tid):
-        with engine.connect() as cn:
+        with engine.begin() as cn:
             sql = task.update().where(task.c.id == tid
             ).values(abort=True)
             cn.execute(sql)

rework_ui/schema.py

@@ -14,10 +14,10 @@ taskstable = Table(
 
 
 def init(engine):
-    with engine.connect() as cn:
+    with engine.begin() as cn:
         taskstable.create(cn)
 
 
 def reset(engine):
-    with engine.connect() as cn:
+    with engine.begin() as cn:
         taskstable.drop(cn, checkfirst=True)

rework_ui/taskstable.py

@@ -35,7 +35,7 @@ def refresh_tasks(engine, inithash, domain):
             domain=domain,
             content=htmltable
         )
-        with engine.connect() as cn:
+        with engine.begin() as cn:
             cn.execute(sql)
         inithash = thash
         # cleanup old tables
@@ -44,7 +44,7 @@ def refresh_tasks(engine, inithash, domain):
         ).where(
             taskstable.c.domain == domain
         )
-        with engine.connect() as cn:
+        with engine.begin() as cn:
             cn.execute(sql)
     return inithash
 
@@ -75,7 +75,7 @@ def refresh_tasks_file(engine, loop=False, sleeptime=2):
 
 
 def tasks_info(engine, domain):
-    with engine.connect() as cn:
+    with engine.begin() as cn:
         sql = select(
             [task.c.id, task.c.status, operation.c.domain]
         ).order_by(desc(task.c.id)

tests/test_rui.py

@@ -126,7 +126,7 @@ def test_task_life_cycle(engine, client, refresh):
 
 
 def test_tasks_table(engine, client, refresh):
-    with engine.connect() as cn:
+    with engine.begin() as cn:
         cn.execute('delete from rework.task')
 
     with workers(engine):