Adds after_job_end hook (#355)

* Adds after_job_end hook

* Fixes test

Author: AngellusMortis

arq/worker.py

@@ -150,6 +150,7 @@ class Worker:
     :param on_shutdown: coroutine function to run at shutdown
     :param on_job_start: coroutine function to run on job start
     :param on_job_end: coroutine function to run on job end
+    :param after_job_end: coroutine function to run after job has ended and results have been recorded
     :param handle_signals: default true, register signal handlers,
       set to false when running inside other async framework
     :param max_jobs: maximum number of jobs to run at a time
@@ -189,6 +190,7 @@ def __init__(
         on_shutdown: Optional['StartupShutdown'] = None,
         on_job_start: Optional['StartupShutdown'] = None,
         on_job_end: Optional['StartupShutdown'] = None,
+        after_job_end: Optional['StartupShutdown'] = None,
         handle_signals: bool = True,
         max_jobs: int = 10,
         job_timeout: 'SecondsTimedelta' = 300,
@@ -227,6 +229,7 @@ def __init__(
         self.on_shutdown = on_shutdown
         self.on_job_start = on_job_start
         self.on_job_end = on_job_end
+        self.after_job_end = after_job_end
         self.sem = asyncio.BoundedSemaphore(max_jobs)
         self.job_timeout_s = to_seconds(job_timeout)
         self.keep_result_s = to_seconds(keep_result)
@@ -635,6 +638,9 @@ async def job_failed(exc: BaseException) -> None:
             )
         )
 
+        if self.after_job_end:
+            await self.after_job_end(ctx)
+
     async def finish_job(
         self,
         job_id: str,

tests/test_worker.py

@@ -920,6 +920,10 @@ async def on_end(ctx):
         assert ctx['job_id'] == 'testing'
         result['called'] += 1
 
+    async def after_end(ctx):
+        assert ctx['job_id'] == 'testing'
+        result['called'] += 2
+
     async def test(ctx):
         return
 
@@ -928,6 +932,7 @@ async def test(ctx):
         functions=[func(test, name='func')],
         on_job_start=on_start,
         on_job_end=on_end,
+        after_job_end=after_end,
         job_timeout=0.2,
         poll_delay=0.1,
     )
@@ -939,7 +944,7 @@ async def test(ctx):
     assert worker.jobs_complete == 1
     assert worker.jobs_failed == 0
     assert worker.jobs_retried == 0
-    assert result['called'] == 2
+    assert result['called'] == 4
 
 
 async def test_worker_timezone_defaults_to_system_timezone(worker):