Returns a dict with stats about all the queues. The keys are the queue names, the values are dicts representing how many tasks are in a given status ("queued", "active", "error" or "scheduled"). Example return value: { "default": { "queued": 1, "error": 2 }
(self)
| 538 | Semaphore.set_system_lock(self.connection, key, timeout) |
| 539 | |
| 540 | def get_queue_stats(self) -> Dict[str, Dict[str, str]]: |
| 541 | """ |
| 542 | Returns a dict with stats about all the queues. The keys are the queue |
| 543 | names, the values are dicts representing how many tasks are in a given |
| 544 | status ("queued", "active", "error" or "scheduled"). |
| 545 | |
| 546 | Example return value: |
| 547 | { "default": { "queued": 1, "error": 2 } } |
| 548 | """ |
| 549 | |
| 550 | states = (QUEUED, ACTIVE, SCHEDULED, ERROR) |
| 551 | |
| 552 | pipeline = self.connection.pipeline() |
| 553 | for state in states: |
| 554 | pipeline.smembers(self._key(state)) |
| 555 | queue_results = pipeline.execute() |
| 556 | |
| 557 | pipeline = self.connection.pipeline() |
| 558 | for state, result in zip(states, queue_results): |
| 559 | for queue in result: |
| 560 | pipeline.zcard(self._key(state, queue)) |
| 561 | card_results = pipeline.execute() |
| 562 | |
| 563 | queue_stats: Dict[str, Dict[str, str]] = defaultdict(dict) |
| 564 | for state, result in zip(states, queue_results): |
| 565 | for queue in result: |
| 566 | queue_stats[queue][state] = card_results.pop(0) |
| 567 | |
| 568 | return queue_stats |
| 569 | |
| 570 | def purge_errored_tasks( |
| 571 | self, |