(self, task_id)
| 210 | f.flush() |
| 211 | |
| 212 | def handle_task_start(self, task_id): |
| 213 | maxkb_logger.debug('handle_task_start') |
| 214 | log_path = get_celery_task_log_path(task_id) |
| 215 | thread_id = self.get_current_thread_id() |
| 216 | self.task_id_thread_id_mapper[task_id] = thread_id |
| 217 | f = open(log_path, 'ab') |
| 218 | self.thread_id_fd_mapper[thread_id] = f |
| 219 | |
| 220 | def handle_task_end(self, task_id): |
| 221 | maxkb_logger.debug('handle_task_end') |
nothing calls this directly
no test coverage detected