(self, item_id, history_result,
status: Optional['PromptQueue.ExecutionStatus'], process_item=None)
| 1272 | messages: List[str] |
| 1273 | |
| 1274 | def task_done(self, item_id, history_result, |
| 1275 | status: Optional['PromptQueue.ExecutionStatus'], process_item=None): |
| 1276 | with self.mutex: |
| 1277 | prompt = self.currently_running.pop(item_id) |
| 1278 | if len(self.history) > MAXIMUM_HISTORY_SIZE: |
| 1279 | self.history.pop(next(iter(self.history))) |
| 1280 | |
| 1281 | status_dict: Optional[dict] = None |
| 1282 | if status is not None: |
| 1283 | status_dict = copy.deepcopy(status._asdict()) |
| 1284 | |
| 1285 | if process_item is not None: |
| 1286 | prompt = process_item(prompt) |
| 1287 | |
| 1288 | self.history[prompt[1]] = { |
| 1289 | "prompt": prompt, |
| 1290 | "outputs": {}, |
| 1291 | 'status': status_dict, |
| 1292 | } |
| 1293 | self.history[prompt[1]].update(history_result) |
| 1294 | self.server.queue_updated() |
| 1295 | |
| 1296 | # Note: slow |
| 1297 | def get_current_queue(self): |
no test coverage detected