feat(job): add log method [python] (#2476) ref #2472

python/bullmq/job.py

@@ -189,6 +189,9 @@ async def moveToFailed(self, err, token:str, fetchNext:bool = False):
 
         self.attemptsMade = self.attemptsMade + 1
 
+    def log(self, logRow: str):
+        return Job.addJobLog(self.queue, self.id, logRow, self.opts.get("keepLogs", 0))
+
     async def saveStacktrace(self, pipe, err:str):
         stacktrace = traceback.format_exc()
         stackTraceLimit = self.opts.get("stackTraceLimit")
@@ -261,6 +264,19 @@ async def fromId(queue: Queue, jobId: str):
         if len(raw_data):
             return Job.fromJSON(queue, raw_data, jobId)
 
+    @staticmethod
+    async def addJobLog(queue: Queue, jobId: str, logRow: str, keepLogs: int = 0):
+        logs_key = f"{queue.prefix}:{queue.name}:{jobId}:logs"
+        multi = await queue.client.pipeline()
+
+        multi.rpush(logs_key, logRow)
+
+        if keepLogs:
+            multi.ltrim(logs_key, -keepLogs, -1)
+
+        result = await multi.execute()
+
+        return min(keepLogs, result[0]) if keepLogs else result[0]
 
 def optsFromJSON(rawOpts: dict) -> dict:
     # opts = json.loads(rawOpts)

python/run_tests.sh

@@ -1,8 +1,3 @@
 #!/bin/bash
 redis-cli flushall
-python3 -m unittest -v tests.bulk_tests
-python3 -m unittest -v tests.delay_tests
-python3 -m unittest -v tests.flow_tests
 python3 -m unittest -v tests.job_tests
-python3 -m unittest -v tests.queue_tests
-python3 -m unittest -v tests.worker_tests

python/tests/job_tests.py

@@ -48,6 +48,18 @@ async def test_get_job_state(self):
 
         await queue.close()
 
+    async def test_job_log(self):
+        queue = Queue(queueName)
+        firstLog = 'some log text 1'
+        secondLog = 'some log text 2'
+        job = await queue.add("test-job", {"foo": "bar"}, {})
+        await job.log(firstLog)
+        log_count = await job.log(secondLog)
+
+        self.assertEqual(log_count, 2)
+
+        await queue.close()
+
     async def test_update_job_data(self):
         queue = Queue(queueName)
         job = await queue.add("test", {"foo": "bar"}, {})