daniel/oshipka
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix after runobeat

Browse Source
This commit is contained in:
Daniel Tsvetkov 2020-03-24 11:55:54 +01:00
parent a486cd7f66
commit 4a8d8da076
3 changed files with 23 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
bootstrap/.gitignore vendored
View File

@ -1,5 +1,5 @@
.idea .idea
venv venv
*.pyc *.pyc
data/db.sqlite data/
__pycache__ __pycache__

18
oshipka/webapp/async_routes.py
View File

@ -14,8 +14,11 @@ from oshipka.webapp import test_bp, oshipka_bp
TASKS = {} TASKS = {}
def register_task(task_name, task_func, *args, **kwargs): def register_task(task_name, task_func, package, *args, **kwargs):
TASKS[task_name] = task_func TASKS[task_name] = {
"func": task_func,
"package": package,
}
def stateful_task(*args, **kwargs): def stateful_task(*args, **kwargs):
@ -25,7 +28,7 @@ def stateful_task(*args, **kwargs):
time.sleep(1) time.sleep(1)
register_task("stateful_task", stateful_task) register_task("stateful_task", stateful_task, "oshipka.webapp.async_routes.{}")
class Task(db.Model): class Task(db.Model):
@ -71,7 +74,9 @@ def tail(filename):
yield "data: {}\n\n".format(line) yield "data: {}\n\n".format(line)
def worker_start_task(task_name, func_name, task_kwargs): def worker_start_task(task_name, task_kwargs):
reg_task = TASKS.get(task_name)
func_name = reg_task.get('package').format(reg_task.get('func').__name__)
uuid = str(uuid4()) uuid = str(uuid4())
task = Task(name=task_name, task = Task(name=task_name,
uuid=uuid, uuid=uuid,
@ -86,9 +91,8 @@ def worker_start_task(task_name, func_name, task_kwargs):
@test_bp.route('/tasks/<task_name>/start', methods=['GET', 'POST']) @test_bp.route('/tasks/<task_name>/start', methods=['GET', 'POST'])
def start_task(task_name): def start_task(task_name):
task_kwargs = {k: v for k, v in request.form.items() if k != 'csrf_token'} task_kwargs = {k: v for k, v in request.form.items() if k != 'csrf_token'}
func_name = "oshipka.webapp.async_routes.{}".format(TASKS.get(task_name).__name__) task_uuid = worker_start_task(task_name, task_kwargs)
async_task_id = worker_start_task(task_name, func_name, task_kwargs) return redirect(url_for('test_bp.get_task_status', task_uuid=task_uuid))
return redirect(url_for('test_bp.get_task_status', task_uuid=async_task_id))
def get_task_ctx(task_uuid): def get_task_ctx(task_uuid):

11
oshipka/worker.py
View File

@ -73,11 +73,17 @@ class DirectoryWatcherHandler(FileSystemEventHandler):
open(lock_fname, "w").write("done") open(lock_fname, "w").write("done")
os.remove(lock_fname) os.remove(lock_fname)
print("WATCHER: Released lock for {}.".format(task_uuid)) print("WATCHER: Released lock for {}.".format(task_uuid))
except Timeout as e: except Timeout:
return
except OSError:
return return
except Exception as e: except Exception as e:
print("WATCHER: Exception: {}".format(e)) print("WATCHER: Exception: {}".format(e))
finally: finally:
try:
os.remove(lock_fname)
except Exception:
...
lock.release() lock.release()
@ -135,7 +141,10 @@ class Worker(object):
print("Worker {} started task: {}".format(self.worker_name, task.name)) print("Worker {} started task: {}".format(self.worker_name, task.name))
task_func = dyn_import(task.func_name) task_func = dyn_import(task.func_name)
self.before_task(task.uuid) self.before_task(task.uuid)
try:
task_func(*json.loads(task.args), **json.loads(task.kwargs)) task_func(*json.loads(task.args), **json.loads(task.kwargs))
except Exception as e:
print("WORKER: Exception: {}".format(e))
self.after_task(task.uuid) self.after_task(task.uuid)
print("Worker {} finished task: {}".format(self.worker_name, task.name)) print("Worker {} finished task: {}".format(self.worker_name, task.name))