Updated on_task_process_exception interface; replaced "function" with "name" in SimpleJob processing

Author: WeirdCarrotMonster

tests/test_simple_job_factory.py

@@ -24,12 +24,12 @@ async def run(self, **kwargs):
     task_a = Task(
         "test",
         timeout=0,
-        encoded_data=json.dumps({"kwargs": {"test": 1}, "function": "job_a"}),
+        encoded_data=json.dumps({"kwargs": {"test": 1}, "name": "job_a"}),
     )
     task_b = Task(
         "test",
         timeout=0,
-        encoded_data=json.dumps({"kwargs": {"test": 1}, "function": "job_b"}),
+        encoded_data=json.dumps({"kwargs": {"test": 1}, "name": "job_b"}),
     )
 
     factory = SimpleJobFactory(handlers=handlers)
@@ -45,7 +45,7 @@ def test_missing_key_job_creation():
     task = Task(
         "test",
         timeout=0,
-        encoded_data=json.dumps({"kwargs": {"test": 1}, "function": "job"}),
+        encoded_data=json.dumps({"kwargs": {"test": 1}, "name": "job"}),
     )
     factory = SimpleJobFactory(handlers={})
 

tests/test_worker.py

@@ -33,7 +33,7 @@ async def redis_connection(cls):
     run_task = asyncio.create_task(run_coro)
     await asyncio.wait_for(worker.started.wait(), timeout=1)
 
-    scheduled_task = await task_queue.add_task({"function": "job", "kwargs": {}})
+    scheduled_task = await task_queue.add_task({"name": "job", "kwargs": {}})
 
     await asyncio.wait_for(event.wait(), timeout=1)
     await worker.stop()
@@ -69,7 +69,7 @@ async def redis_connection(cls):
     run_task = asyncio.create_task(run_coro)
     await asyncio.wait_for(worker.started.wait(), timeout=1)
 
-    scheduled_task = await task_queue.add_task({"function": "job", "kwargs": {}})
+    scheduled_task = await task_queue.add_task({"name": "job", "kwargs": {}})
 
     await asyncio.wait_for(event.wait(), timeout=1)
     await worker.stop()
@@ -103,7 +103,7 @@ async def redis_connection(cls):
     run_task = asyncio.create_task(run_coro)
     await asyncio.wait_for(worker.started.wait(), timeout=1)
 
-    scheduled_task = await task_queue.add_task({"function": "job", "kwargs": {}})
+    scheduled_task = await task_queue.add_task({"name": "job", "kwargs": {}})
 
     await asyncio.wait_for(event.wait(), timeout=1)
     await asyncio.wait_for(worker.completed_task.wait(), timeout=1)
@@ -132,8 +132,8 @@ async def redis_connection(cls):
 
     worker = build_worker(redis_connection, Settings, [task_queue.name], max_jobs=1)
 
-    scheduled_task_1 = await task_queue.add_task({"function": "job", "kwargs": {}})
-    scheduled_task_2 = await task_queue.add_task({"function": "job", "kwargs": {}})
+    scheduled_task_1 = await task_queue.add_task({"name": "job", "kwargs": {}})
+    scheduled_task_2 = await task_queue.add_task({"name": "job", "kwargs": {}})
 
     run_coro = worker.run()
     run_task = asyncio.create_task(run_coro)
@@ -171,7 +171,7 @@ async def redis_connection(cls):
     run_task = asyncio.create_task(run_coro)
     await asyncio.wait_for(worker.started.wait(), timeout=1)
 
-    scheduled_task = await task_queue.add_task({"function": "job", "kwargs": {}})
+    scheduled_task = await task_queue.add_task({"name": "job", "kwargs": {}})
     await asyncio.wait_for(worker.got_task.wait(), timeout=1)
 
     await worker.stop()
@@ -208,7 +208,7 @@ async def redis_connection(cls):
     run_task = asyncio.create_task(run_coro)
     await asyncio.wait_for(worker.started.wait(), timeout=1)
 
-    scheduled_task = await task_queue.add_task({"function": "job", "kwargs": {}})
+    scheduled_task = await task_queue.add_task({"name": "job", "kwargs": {}})
     await asyncio.wait_for(worker.got_task.wait(), timeout=1)
 
     await worker.stop()
@@ -240,7 +240,7 @@ async def redis_connection(cls):
     run_task = asyncio.create_task(run_coro)
     await asyncio.wait_for(worker.started.wait(), timeout=1)
 
-    scheduled_task = await task_queue.add_task({"function": "job", "kwargs": {}})
+    scheduled_task = await task_queue.add_task({"name": "job", "kwargs": {}})
     await asyncio.wait_for(worker.got_task.wait(), timeout=1)
 
     await worker.stop()
@@ -275,7 +275,7 @@ async def redis_connection(cls):
     run_task = asyncio.create_task(run_coro)
     await asyncio.wait_for(worker.started.wait(), timeout=1)
 
-    scheduled_task = await task_queue.add_task({"function": "job", "kwargs": {}})
+    scheduled_task = await task_queue.add_task({"name": "job", "kwargs": {}})
     await asyncio.wait_for(worker.got_task.wait(), timeout=1)
 
     await worker.stop()
@@ -308,7 +308,7 @@ async def redis_connection(cls):
     await asyncio.wait_for(worker.started.wait(), timeout=1)
 
     scheduled_task = await task_queue.add_task(
-        {"function": "job", "kwargs": {}}, retry_policy=RetryPolicy.LINEAR
+        {"name": "job", "kwargs": {}}, retry_policy=RetryPolicy.LINEAR
     )
     await asyncio.wait_for(worker.got_task.wait(), timeout=1)
 
@@ -332,7 +332,7 @@ async def redis_connection(cls):
         factory_kwargs = {"handlers": {}}
 
     scheduled_task = await task_queue.add_task(
-        {"function": "job", "kwargs": {}}, task_timeout=0
+        {"name": "job", "kwargs": {}}, task_timeout=0
     )
     await task_queue.get_task()
 
@@ -359,7 +359,7 @@ async def redis_connection(cls):
             return redis_connection
 
         @staticmethod
-        async def on_task_process_exception(exc_info):
+        async def on_task_process_exception(job, exc_info):
             exception_handler()
 
         queue_namespace = task_queue.namespace
@@ -371,7 +371,7 @@ async def on_task_process_exception(exc_info):
     run_task = asyncio.create_task(run_coro)
     await asyncio.wait_for(worker.started.wait(), timeout=1)
 
-    scheduled_task = await task_queue.add_task({"function": "job", "kwargs": {}})
+    scheduled_task = await task_queue.add_task({"name": "job", "kwargs": {}})
     await asyncio.wait_for(worker.got_task.wait(), timeout=1)
 
     await worker.stop()
@@ -411,7 +411,7 @@ async def on_task_process_exception(exc_info):
     run_task = asyncio.create_task(run_coro)
     await asyncio.wait_for(worker.started.wait(), timeout=1)
 
-    scheduled_task = await task_queue.add_task({"function": "job", "kwargs": {}})
+    scheduled_task = await task_queue.add_task({"name": "job", "kwargs": {}})
     await asyncio.wait_for(worker.got_task.wait(), timeout=1)
 
     await worker.stop()

yatq/worker/factory/simple.py

@@ -22,7 +22,7 @@ def create_job(self, task: "Task") -> T_SimpleJobClass:
         if not task_data:
             raise ValueError("Task data is not set")
 
-        task_function = task_data["function"]
+        task_function = task_data["name"]
         handler_class = self.handlers[task_function]
 
         return handler_class(task)

yatq/worker/runner.py

@@ -12,6 +12,7 @@
 from yatq.exceptions import TaskRescheduleException
 from yatq.queue import Queue
 from yatq.worker.factory.base import BaseJobFactory
+from yatq.worker.job.base import BaseJob
 from yatq.worker.worker_settings import T_ExcInfo, WorkerSettings
 
 LOGGER = logging.getLogger("yatq.worker")
@@ -24,7 +25,7 @@ def __init__(
         self,
         queue_list: List[Queue],
         task_factory: BaseJobFactory,
-        on_task_process_exception: Callable[[T_ExcInfo], Awaitable],
+        on_task_process_exception: Callable[[BaseJob, T_ExcInfo], Awaitable],
         poll_interval: float = 2.0,
         max_jobs: int = 8,
         gravekeeper_interval: float = 30.0,
@@ -142,7 +143,7 @@ async def _handle_task(self, wrapper, queue) -> None:
 
             exc_info = cast(T_ExcInfo, sys.exc_info())
             try:
-                await self._on_task_process_exception(exc_info)
+                await self._on_task_process_exception(task_job, exc_info)
             except Exception:
                 LOGGER.exception("Exception in exception handler")
 

yatq/worker/worker_settings.py

@@ -4,6 +4,7 @@
 import aioredis
 
 from yatq.worker.factory.simple import SimpleJobFactory
+from yatq.worker.job.simple import SimpleJob
 
 T_ExcInfo = Tuple[Type[BaseException], BaseException, TracebackType]
 
@@ -29,6 +30,7 @@ async def redis_client() -> aioredis.Redis:  # pragma: no cover
 
     @staticmethod
     async def on_task_process_exception(
+        job: SimpleJob,
         exc_info: T_ExcInfo,
     ) -> None:  # pragma: no cover
         ...