(self, task: "Task", hard_timeout: float)
| 61 | """ |
| 62 | |
| 63 | def run_single_task(self, task: "Task", hard_timeout: float) -> None: |
| 64 | with UnixSignalDeathPenalty(hard_timeout): |
| 65 | task.func(*task.args, **task.kwargs) |
| 66 | |
| 67 | def run_batch_tasks(self, tasks: List["Task"], hard_timeout: float) -> None: |
| 68 | params = [{"args": task.args, "kwargs": task.kwargs} for task in tasks] |
nothing calls this directly
no test coverage detected