(
self,
request: CompletionRequest,
)
| 15565 | self.condition.wait(timeout=0.01) |
| 15566 | |
| 15567 | def submit_request( |
| 15568 | self, |
| 15569 | request: CompletionRequest, |
| 15570 | ) -> Tuple[CompletionStream, Callable[[], None]]: |
| 15571 | mailbox: "queue.Queue[object]" = queue.Queue() |
| 15572 | done = threading.Event() |
| 15573 | cancelled = threading.Event() |
| 15574 | sentinel = object() |
| 15575 | result_box: Dict[str, OpenAICompletion] = {} |
| 15576 | error_box: Dict[str, BaseException] = {} |
| 15577 | |
| 15578 | def on_stream_chunk(chunk: CompletionChunk) -> None: |
| 15579 | mailbox.put(chunk) |
| 15580 | |
| 15581 | def on_done(result: OpenAICompletion) -> None: |
| 15582 | result_box["result"] = result |
| 15583 | done.set() |
| 15584 | mailbox.put(sentinel) |
| 15585 | |
| 15586 | def on_error(exc: BaseException) -> None: |
| 15587 | error_box["error"] = exc |
| 15588 | done.set() |
| 15589 | mailbox.put(sentinel) |
| 15590 | |
| 15591 | request.on_stream_chunk = on_stream_chunk |
| 15592 | request.on_done = on_done |
| 15593 | request.on_error = on_error |
| 15594 | |
| 15595 | def cancel() -> None: |
| 15596 | if cancelled.is_set(): |
| 15597 | return |
| 15598 | cancelled.set() |
| 15599 | try: |
| 15600 | def cancel_request() -> None: |
| 15601 | self.scheduler.cancel(request.id) |
| 15602 | |
| 15603 | self.enqueue(cancel_request) |
| 15604 | except RuntimeError: |
| 15605 | pass |
| 15606 | |
| 15607 | def stream() -> CompletionStream: |
| 15608 | try: |
| 15609 | while True: |
| 15610 | item = mailbox.get() |
| 15611 | if item is sentinel: |
| 15612 | break |
| 15613 | yield cast(CompletionChunk, item) |
| 15614 | finally: |
| 15615 | if not done.is_set(): |
| 15616 | cancel() |
| 15617 | if "error" in error_box: |
| 15618 | raise error_box["error"] |
| 15619 | result = result_box.get("result") |
| 15620 | if result is None: |
| 15621 | raise RuntimeError("missing completion result") |
| 15622 | return result |
| 15623 | |
| 15624 | def submit_request() -> None: |
no test coverage detected