Simplify job context

docs/reference.rst

@@ -32,7 +32,7 @@ When tasks are created with argument ``pass_context``, they are provided a
 `JobContext` argument:
 
 .. autoclass:: procrastinate.JobContext
-    :members: app, worker_name, worker_queues, job, task
+    :members: app, worker_name, worker_queues, job
 
 Blueprints
 ----------

procrastinate/job_context.py

@@ -6,29 +6,27 @@
 import attr
 
 from procrastinate import app as app_module
-from procrastinate import jobs, tasks, utils
+from procrastinate import jobs, utils
 
 
 @attr.dataclass(kw_only=True)
 class JobResult:
-    start_timestamp: float | None = None
+    start_timestamp: float
     end_timestamp: float | None = None
     result: Any = None
 
     def duration(self, current_timestamp: float) -> float | None:
-        if self.start_timestamp is None:
-            return None
         return (self.end_timestamp or current_timestamp) - self.start_timestamp
 
     def as_dict(self):
         result = {}
-        if self.start_timestamp:
-            result.update(
-                {
-                    "start_timestamp": self.start_timestamp,
-                    "duration": self.duration(current_timestamp=time.time()),
-                }
-            )
+        result.update(
+            {
+                "start_timestamp": self.start_timestamp,
+                "duration": self.duration(current_timestamp=time.time()),
+            }
+        )
+
         if self.end_timestamp:
             result.update({"end_timestamp": self.end_timestamp, "result": self.result})
         return result
@@ -48,11 +46,10 @@ class JobContext:
     worker_queues: Iterable[str] | None = None
     #: Corresponding :py:class:`~jobs.Job`
     job: jobs.Job
-    #: Corresponding :py:class:`~tasks.Task`
-    task: tasks.Task | None = None
-    job_result: JobResult = attr.ib(factory=JobResult)
+    #: Time the job started to be processed
+    start_timestamp: float
+
     additional_context: dict = attr.ib(factory=dict)
-    task_result: Any = None
 
     should_abort: Callable[[], bool]
 
@@ -62,11 +59,3 @@ def evolve(self, **update: Any) -> JobContext:
     @property
     def queues_display(self) -> str:
         return utils.queues_display(self.worker_queues)
-
-    def job_description(self, current_timestamp: float) -> str:
-        message = f"worker: {self.job.call_string}"
-        duration = self.job_result.duration(current_timestamp)
-        if duration is not None:
-            message += f" (started {duration:.3f} s ago)"
-
-        return message

procrastinate/worker.py

@@ -79,7 +79,9 @@ def stop(self):
             return
         self.logger.info(
             "Stop requested",
-            extra=self._log_extra(context=None, action="stopping_worker"),
+            extra=self._log_extra(
+                context=None, action="stopping_worker", job_result=None
+            ),
         )
 
         self._stop_event.set()
@@ -98,7 +100,11 @@ def find_task(self, task_name: str) -> tasks.Task:
             raise exceptions.TaskNotFound from exc
 
     def _log_extra(
-        self, action: str, context: job_context.JobContext | None, **kwargs: Any
+        self,
+        action: str,
+        context: job_context.JobContext | None,
+        job_result: job_context.JobResult | None,
+        **kwargs: Any,
     ) -> types.JSONDict:
         extra: types.JSONDict = {
             "action": action,
@@ -113,7 +119,7 @@ def _log_extra(
 
         return {
             **extra,
-            **(context.job_result if context else job_context.JobResult()).as_dict(),
+            **(job_result.as_dict() if job_result else {}),
             **kwargs,
         }
 
@@ -123,6 +129,7 @@ async def _persist_job_status(
         status: jobs.Status,
         retry_decision: retry.RetryDecision | None,
         context: job_context.JobContext,
+        job_result: job_context.JobResult | None,
     ):
         if retry_decision:
             await self.app.job_manager.retry_job(
@@ -146,13 +153,19 @@ async def _persist_job_status(
 
         self.logger.debug(
             f"Acknowledged job completion {job.call_string}",
-            extra=self._log_extra(action="finish_task", context=context, status=status),
+            extra=self._log_extra(
+                action="finish_task",
+                context=context,
+                status=status,
+                job_result=job_result,
+            ),
         )
 
     def _log_job_outcome(
         self,
         status: jobs.Status,
         context: job_context.JobContext,
+        job_result: job_context.JobResult | None,
         job_retry: exceptions.JobRetry | None,
         exc_info: bool | BaseException = False,
     ):
@@ -168,43 +181,46 @@ def _log_job_outcome(
         text = f"Job {context.job.call_string} ended with status: {log_title}, "
         # in practice we should always have a start and end timestamp here
         # but in theory the JobResult class allows it to be None
-        if context.job_result.start_timestamp and context.job_result.end_timestamp:
-            duration = (
-                context.job_result.end_timestamp - context.job_result.start_timestamp
-            )
+        if job_result and job_result.start_timestamp and job_result.end_timestamp:
+            duration = job_result.end_timestamp - job_result.start_timestamp
             text += f"lasted {duration:.3f} s"
-        if context.job_result.result:
-            text += f" - Result: {context.job_result.result}"[:250]
+        if job_result and job_result.result:
+            text += f" - Result: {job_result.result}"[:250]
 
-        extra = self._log_extra(context=context, action=log_action)
+        extra = self._log_extra(
+            context=context, action=log_action, job_result=job_result
+        )
         log_level = logging.ERROR if status == jobs.Status.FAILED else logging.INFO
         logger.log(log_level, text, extra=extra, exc_info=exc_info)
 
     async def _process_job(self, context: job_context.JobContext):
         """
         Processes a given job and persists its status
         """
-        task = context.task
+        task = self.app.tasks.get(context.job.task_name)
         job_retry = None
         exc_info = False
         retry_decision = None
         job = context.job
 
-        job_result = context.job_result
-        job_result.start_timestamp = time.time()
+        job_result = job_context.JobResult(start_timestamp=context.start_timestamp)
 
         try:
             if not task:
                 raise exceptions.TaskNotFound
 
             self.logger.debug(
                 f"Loaded job info, about to start job {job.call_string}",
-                extra=self._log_extra(context=context, action="loaded_job_info"),
+                extra=self._log_extra(
+                    context=context, action="loaded_job_info", job_result=job_result
+                ),
             )
 
             self.logger.info(
                 f"Starting job {job.call_string}",
-                extra=self._log_extra(context=context, action="start_job"),
+                extra=self._log_extra(
+                    context=context, action="start_job", job_result=job_result
+                ),
             )
 
             exc_info: bool | BaseException = False
@@ -248,6 +264,7 @@ async def ensure_async() -> Callable[..., Awaitable]:
                             context=context,
                             action="task_not_found",
                             exception=str(e),
+                            job_result=job_result,
                         ),
                     )
         finally:
@@ -265,6 +282,7 @@ async def ensure_async() -> Callable[..., Awaitable]:
             self._log_job_outcome(
                 status=status,
                 context=context,
+                job_result=job_result,
                 job_retry=job_retry,
                 exc_info=exc_info,
             )
@@ -275,6 +293,7 @@ async def ensure_async() -> Callable[..., Awaitable]:
                     status=status,
                     retry_decision=retry_decision,
                     context=context,
+                    job_result=job_result,
                 )
             )
             try:
@@ -312,8 +331,8 @@ async def _fetch_and_process_jobs(self):
                 if self.additional_context
                 else {},
                 job=job,
-                task=self.app.tasks.get(job.task_name),
                 should_abort=lambda: job_id in self._job_ids_to_abort,
+                start_timestamp=time.time(),
             )
             job_task = asyncio.create_task(
                 self._process_job(context),
@@ -397,16 +416,18 @@ def _abort_job(
         self._job_ids_to_abort.add(context.job.id)
 
         log_message: str
-        if not context.task:
+        task = self.app.tasks.get(context.job.task_name)
+        if not task:
             log_message = "Received a request to abort a job but the job has no associated task. No action to perform"
-        elif not asyncio.iscoroutinefunction(context.task.func):
+        elif not asyncio.iscoroutinefunction(task.func):
             log_message = "Received a request to abort a synchronous job. Job is responsible for aborting by checking context.should_abort"
         else:
             log_message = "Received a request to abort an asynchronous job. Cancelling asyncio task"
             process_job_task.cancel()
 
         self.logger.debug(
-            log_message, extra=self._log_extra(action="abort_job", context=context)
+            log_message,
+            extra=self._log_extra(action="abort_job", context=context, job_result=None),
         )
 
     async def _shutdown(self, side_tasks: list[asyncio.Task]):
@@ -418,10 +439,12 @@ async def _shutdown(self, side_tasks: list[asyncio.Task]):
 
         now = time.time()
         for context in self._running_jobs.values():
+            duration = now - context.start_timestamp
             self.logger.info(
-                "Waiting for job to finish: "
-                + context.job_description(current_timestamp=now),
-                extra=self._log_extra(context=None, action="ending_job"),
+                f"Waiting for job to finish: worker: {context.job.call_string} (started {duration:.3f} s ago)",
+                extra=self._log_extra(
+                    context=None, action="ending_job", job_result=None
+                ),
             )
 
         # wait for any in progress job to complete processing
@@ -430,7 +453,7 @@ async def _shutdown(self, side_tasks: list[asyncio.Task]):
         self.logger.info(
             f"Stopped worker on {utils.queues_display(self.queues)}",
             extra=self._log_extra(
-                action="stop_worker", queues=self.queues, context=None
+                action="stop_worker", queues=self.queues, context=None, job_result=None
             ),
         )
 
@@ -455,7 +478,7 @@ async def _run_loop(self):
         self.logger.info(
             f"Starting worker on {utils.queues_display(self.queues)}",
             extra=self._log_extra(
-                action="start_worker", context=None, queues=self.queues
+                action="start_worker", context=None, queues=self.queues, job_result=None
             ),
         )
         self._new_job_event.clear()
@@ -480,6 +503,7 @@ async def _run_loop(self):
                             context=None,
                             action="stop_worker",
                             queues=self.queues,
+                            job_result=None,
                         ),
                     )
                     self._stop_event.set()

tests/unit/test_builtin_tasks.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import time
 from typing import cast
 
 from procrastinate import builtin_tasks, job_context
@@ -10,7 +11,9 @@
 async def test_remove_old_jobs(app: App, job_factory):
     job = job_factory()
     await builtin_tasks.remove_old_jobs(
-        job_context.JobContext(app=app, job=job, should_abort=lambda: False),
+        job_context.JobContext(
+            app=app, job=job, should_abort=lambda: False, start_timestamp=time.time()
+        ),
         max_hours=2,
         queue="queue_a",
         remove_failed=True,

tests/unit/test_job_context.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import time
+
 import pytest
 
 from procrastinate import job_context
@@ -9,7 +11,6 @@
 @pytest.mark.parametrize(
     "job_result, expected",
     [
-        (job_context.JobResult(), None),
         (job_context.JobResult(start_timestamp=10), 20),
         (job_context.JobResult(start_timestamp=10, end_timestamp=15), 5),
     ],
@@ -21,7 +22,6 @@ def test_job_result_duration(job_result, expected):
 @pytest.mark.parametrize(
     "job_result, expected",
     [
-        (job_context.JobResult(), {}),
         (
             job_context.JobResult(start_timestamp=10),
             {
@@ -48,26 +48,10 @@ def test_job_result_as_dict(job_result, expected, mocker):
 def test_evolve(app: App, job_factory):
     job = job_factory()
     context = job_context.JobContext(
-        app=app, job=job, worker_name="a", should_abort=lambda: False
-    )
-    assert context.evolve(worker_name="b").worker_name == "b"
-
-
-def test_job_description_job_no_time(app: App, job_factory):
-    job = job_factory(task_name="some_task", id=12, task_kwargs={"a": "b"})
-    descr = job_context.JobContext(
-        worker_name="a", job=job, app=app, should_abort=lambda: False
-    ).job_description(current_timestamp=0)
-    assert descr == "worker: some_task[12](a='b')"
-
-
-def test_job_description_job_time(app: App, job_factory):
-    job = job_factory(task_name="some_task", id=12, task_kwargs={"a": "b"})
-    descr = job_context.JobContext(
-        worker_name="a",
-        job=job,
+        start_timestamp=time.time(),
         app=app,
-        job_result=job_context.JobResult(start_timestamp=20.0),
+        job=job,
+        worker_name="a",
         should_abort=lambda: False,
-    ).job_description(current_timestamp=30.0)
-    assert descr == "worker: some_task[12](a='b') (started 10.000 s ago)"
+    )
+    assert context.evolve(worker_name="b").worker_name == "b"