feat(queue): add clean method [python] (#2194)

python/bullmq/queue.py

@@ -176,6 +176,17 @@ async def getJobCounts(self, *types):
             counts[current_types[index]] = val or 0
         return counts
 
+    async def clean(self, grace: int, limit: int, type: str):
+        """
+        Cleans jobs from a queue. Similar to drain but keeps jobs within a certain
+        grace period
+        
+        * @returns: Id jobs from the deleted records
+        """
+        jobs = await self.scripts.cleanJobsInSet(type, grace, limit)
+
+        return jobs
+
     def getJobState(self, job_id: str):
         return self.scripts.getState(job_id)
 

python/bullmq/scripts.py

@@ -33,6 +33,7 @@ def __init__(self, prefix: str, queueName: str, redisConnection: RedisConnection
         self.commands = {
             "addJob": self.redisClient.register_script(self.getScript("addJob-9.lua")),
             "changePriority": self.redisClient.register_script(self.getScript("changePriority-5.lua")),
+            "cleanJobsInSet": self.redisClient.register_script(self.getScript("cleanJobsInSet-2.lua")),
             "extendLock": self.redisClient.register_script(self.getScript("extendLock-2.lua")),
             "getCounts": self.redisClient.register_script(self.getScript("getCounts-1.lua")),
             "getRanges": self.redisClient.register_script(self.getScript("getRanges-1.lua")),
@@ -116,6 +117,17 @@ def addParentJob(self, job: Job, waiting_children_key: str, pipe = None):
 
         return self.commands["addJob"](keys=keys, args=args, client=pipe)
 
+    def cleanJobsInSetArgs(self, set: str, grace: int, limit:int = 0):
+        keys = [self.toKey(set),
+                self.keys['events']]
+        args = [self.keys[''], round(time.time() * 1000) - grace, limit, set]
+
+        return (keys, args)
+
+    def cleanJobsInSet(self, set: str, grace: int = 0, limit:int = 0):
+        keys, args = self.cleanJobsInSetArgs(set, grace, limit)
+        return self.commands["cleanJobsInSet"](keys=keys, args=args)
+
     def moveToWaitingChildrenArgs(self, job_id, token, opts):
         keys = [self.toKey(job_id) + ":lock",
                 self.keys['active'],