diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 46da185d4..cbabc54ca 100644 --- a/queue_job/controllers/main.py +++ b/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 diff --git a/queue_job/delay.py b/queue_job/delay.py index 60e98dc9d..afb3adcb8 100644 --- a/queue_job/delay.py +++ b/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 diff --git a/queue_job/job.py b/queue_job/job.py index 86314499b..98c442d50 100644 --- a/queue_job/job.py +++ b/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,7 @@ def _load_from_db_record(cls, job_db_record): description=stored.name, channel=stored.channel, identity_key=stored.identity_key, + on_fail_method=stored.on_fail_method_name, ) if stored.date_created: @@ -365,6 +369,7 @@ def __init__( description=None, channel=None, identity_key=None, + on_fail_method=None, ): """Create a Job @@ -407,6 +412,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 +840,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"" diff --git a/queue_job/models/base.py b/queue_job/models/base.py index 2c71cafc1..5a256c170 100644 --- a/queue_job/models/base.py +++ b/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): diff --git a/queue_job/models/queue_job.py b/queue_job/models/queue_job.py index 88b6b4e77..be03118e2 100644 --- a/queue_job/models/queue_job.py +++ b/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 diff --git a/queue_job/tests/test_run_rob_controller.py b/queue_job/tests/test_run_rob_controller.py index 1a15f4363..60939f35b 100644 --- a/queue_job/tests/test_run_rob_controller.py +++ b/queue_job/tests/test_run_rob_controller.py @@ -1,8 +1,10 @@ # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). +from unittest.mock import patch from odoo.tests.common import TransactionCase from ..controllers.main import RunJobController +from ..exception import JobError from ..job import Job @@ -21,3 +23,19 @@ 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, + ): + RunJobController._runjob(self.env, job) + self.assertEqual(job.state, "failed") + self.assertEqual(mocked_hook.call_count, 1)