(self, tasks: List[Task], execution: Dict)
| 921 | return set(self.connection.sscan_iter(key, match=match, count=100000)) |
| 922 | |
| 923 | def store_task_execution(self, tasks: List[Task], execution: Dict) -> None: |
| 924 | serialized_execution = json.dumps(execution) |
| 925 | |
| 926 | for task in tasks: |
| 927 | executions_key = self._key("task", task.id, "executions") |
| 928 | executions_count_key = self._key("task", task.id, "executions_count") |
| 929 | |
| 930 | pipeline = self.connection.pipeline() |
| 931 | pipeline.incr(executions_count_key) |
| 932 | pipeline.rpush(executions_key, serialized_execution) |
| 933 | |
| 934 | if task.max_stored_executions: |
| 935 | pipeline.ltrim(executions_key, -task.max_stored_executions, -1) |
| 936 | |
| 937 | pipeline.execute() |
| 938 | |
| 939 | def run( |
| 940 | self, |
no test coverage detected