[19.0][IMP] queue_job: Implement on fail hook

queue_job/controllers/main.py

@@ -186,6 +186,7 @@ def retry_postpone(job, message, seconds=None):
                 vals = cls._get_failure_values(job, traceback_txt, orig_exception)
                 job.set_failed(**vals)
                 job.store()
+                job.on_fail_hook(vals)
                 buff.close()
             raise
 

queue_job/delay.py

@@ -438,6 +438,7 @@ class Delayable:
         "description",
         "channel",
         "identity_key",
+        "on_fail_method",
     )
     __slots__ = _properties + (
         "recordset",
@@ -457,6 +458,7 @@ def __init__(
         description=None,
         channel=None,
         identity_key=None,
+        on_fail_method=None,
     ):
         self._graph = DelayableGraph()
         self._graph.add_vertex(self)
@@ -469,6 +471,7 @@ def __init__(
         self.description = description
         self.channel = channel
         self.identity_key = identity_key
+        self.on_fail_method = on_fail_method
 
         self._job_method = None
         self._job_args = ()
@@ -547,6 +550,7 @@ def split(self, size, chain=False):
                 description=self.description,
                 channel=self.channel,
                 identity_key=self.identity_key,
+                on_fail_method=self.on_fail_method,
             )
             # Update the __self__
             delayable._job_method = getattr(recordset, self._job_method.__name__)
@@ -583,6 +587,7 @@ def _build_job(self):
             description=self.description,
             channel=self.channel,
             identity_key=self.identity_key,
+            on_fail_method=self.on_fail_method,
         )
         return self._generated_job
 
@@ -633,6 +638,7 @@ def __init__(
         description=None,
         channel=None,
         identity_key=None,
+        on_fail_method=None,
     ):
         self.delayable = Delayable(
             recordset,
@@ -642,6 +648,7 @@ def __init__(
             description=description,
             channel=channel,
             identity_key=identity_key,
+            on_fail_method=on_fail_method,
         )
 
     @property

queue_job/job.py

@@ -200,6 +200,9 @@ class Job:
         be added to a channel if the existing job with the same key is not yet
         started or executed.
 
+    .. attribute::on_fail_method
+
+        A function to be called if the job is failed and will not be retried.
     """
 
     @classmethod
@@ -298,6 +301,9 @@ def _load_from_db_record(cls, job_db_record):
             description=stored.name,
             channel=stored.channel,
             identity_key=stored.identity_key,
+            on_fail_method=getattr(recordset, stored.on_fail_method_name)
+            if stored.on_fail_method_name
+            else None,
         )
 
         if stored.date_created:
@@ -365,6 +371,7 @@ def __init__(
         description=None,
         channel=None,
         identity_key=None,
+        on_fail_method=None,
     ):
         """Create a Job
 
@@ -407,6 +414,11 @@ def __init__(
         self.method_name = func.__name__
         self.recordset = recordset
 
+        if on_fail_method:
+            if not _is_model_method(on_fail_method):
+                raise TypeError("Job accepts only methods of Models")
+            self.on_fail_method_name = on_fail_method.__name__
+
         self.job_config = (
             self.env["queue.job.function"].sudo().job_config(self.job_function_name)
         )
@@ -830,6 +842,11 @@ def set_failed(self, **kw):
             if v is not None:
                 setattr(self, k, v)
 
+    def on_fail_hook(self, fail_vals):
+        on_fail_func = getattr(self.recordset, self.on_fail_method_name, None)
+        if on_fail_func:
+            on_fail_func(**fail_vals)
+
     def __repr__(self):
         return f"<Job {self.uuid}, priority:{self.priority}>"
 

queue_job/models/base.py

@@ -26,6 +26,7 @@ def with_delay(
         description=None,
         channel=None,
         identity_key=None,
+        on_fail_method=None,
     ):
         """Return a ``DelayableRecordset``
 
@@ -59,6 +60,7 @@ def with_delay(
             description=description,
             channel=channel,
             identity_key=identity_key,
+            on_fail_method=on_fail_method,
         )
 
     def delayable(
@@ -69,6 +71,7 @@ def delayable(
         description=None,
         channel=None,
         identity_key=None,
+        on_fail_method=None,
     ):
         """Return a ``Delayable``
 
@@ -140,6 +143,7 @@ def delayable(
             description=description,
             channel=channel,
             identity_key=identity_key,
+            on_fail_method=on_fail_method,
         )
 
     def _patch_job_auto_delay(self, method_name, context_key=None):

queue_job/models/queue_job.py

@@ -77,6 +77,7 @@ class QueueJob(models.Model):
 
     model_name = fields.Char(string="Model", readonly=True)
     method_name = fields.Char(readonly=True)
+    on_fail_method_name = fields.Char(readonly=True)
     records = JobSerialized(
         string="Record(s)",
         readonly=True,
@@ -490,3 +491,6 @@ def _test_job(
             time.sleep(job_duration)
         if commit_within_job:
             self.env.cr.commit()  # pylint: disable=invalid-commit
+
+    def _test_on_fail_hook(self, **kw):
+        pass

queue_job/tests/test_run_rob_controller.py

@@ -1,12 +1,23 @@
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
+from unittest.mock import patch
 
 from odoo.tests.common import TransactionCase
+from odoo.tools import mute_logger
 
 from ..controllers.main import RunJobController
+from ..exception import JobError
 from ..job import Job
 
 
 class TestRunJobController(TransactionCase):
+    def setUp(cls):
+        super().setUp()
+
+        def _clean_queue_job():
+            cls.env["queue.job"].search([]).unlink()
+
+        cls.addCleanup(_clean_queue_job)
+
     def test_get_failure_values(self):
         method = self.env["res.users"].mapped
         job = Job(method)
@@ -21,3 +32,22 @@ def test_runjob_success(self):
         RunJobController._runjob(self.env, job)
         self.assertEqual(job.state, "done")
         self.assertEqual(job.db_record().state, "done")
+
+    def test_runjob_on_fail_hook(self):
+        job = (
+            self.env["queue.job"]
+            .with_delay(on_fail_method=self.env["queue.job"]._test_on_fail_hook)
+            ._test_job(failure_rate=1)
+        )
+        with (
+            self.assertRaises(JobError),
+            patch(
+                "odoo.addons.queue_job.models.queue_job.QueueJob._test_on_fail_hook"
+            ) as mocked_hook,
+            patch("odoo.addons.queue_job.job.Job.in_temporary_env") as mocked_temp_env,
+            mute_logger("odoo.addons.queue_job.controllers.main"),
+        ):
+            mocked_temp_env.return_value.__enter__.return_value = self.env
+            RunJobController._runjob(self.env, job)
+            self.assertEqual(job.state, "failed")
+            self.assertEqual(mocked_hook.call_count, 1)

test_queue_job/tests/test_autovacuum.py

@@ -51,7 +51,6 @@ def test_autovacuum_multi_channel(self):
         job_60days.write(
             {"channel": channel_60days.complete_name, "date_done": date_done}
         )
-
         self.assertEqual(
             len(self.env["queue.job"].search([("channel", "!=", False)])), 2
         )

test_queue_job/tests/test_delay_mocks.py

@@ -308,6 +308,7 @@ def test_mock_with_delay(self):
                     "eta": 15,
                     "identity_key": identity_exact,
                     "max_retries": 1,
+                    "on_fail_method": None,
                     "priority": 15,
                 },
             )