run: serialize arguments for celery task

Author: kpsherva

invenio_jobs/models.py

@@ -49,7 +49,6 @@ class Job(db.Model, Timestamp):
     active = db.Column(db.Boolean, default=True, nullable=False)
     title = db.Column(db.String(255), nullable=False)
     description = db.Column(db.Text)
-    # default_args = db.Column(JSON, default=lambda: dict(), nullable=True)
     task = db.Column(db.String(255))
     default_queue = db.Column(db.String(64))
     schedule = db.Column(JSON, nullable=True)
@@ -177,6 +176,12 @@ def dump(self):
         dict_run["args"] = serialized_args
         return dict_run
 
+    @property
+    def dict_args(self):
+        """Return run arguments."""
+        _args = self.args
+        return json.loads(_args)
+
 
 class Task:
     """Celery Task model."""
@@ -208,3 +213,4 @@ def all(cls):
     def get(cls, id_):
         """Get registered task by id."""
         return cls(current_jobs.registry.get(id_))
+

invenio_jobs/tasks.py

@@ -33,7 +33,7 @@ def execute_run(self, run_id, kwargs=None):
     task = current_jobs.registry.get(run.job.task).task
     update_run(run, status=RunStatusEnum.RUNNING, started_at=datetime.now(timezone.utc))
     try:
-        result = task.apply(kwargs=run.args, throw=True)
+        result = task.apply(kwargs=run.dict_args, throw=True)
     except SystemExit as e:
         update_run(
             run,