A helper to poll a run status until it reaches a terminal state. More information on Run lifecycles can be found here: https://platform.openai.com/docs/assistants/how-it-works/runs-and-run-steps
(
self,
run_id: str,
thread_id: str,
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = not_given,
poll_interval_ms: int | Omit = omit,
)
| 1051 | |
| 1052 | @typing_extensions.deprecated("The Assistants API is deprecated in favor of the Responses API") |
| 1053 | def poll( |
| 1054 | self, |
| 1055 | run_id: str, |
| 1056 | thread_id: str, |
| 1057 | extra_headers: Headers | None = None, |
| 1058 | extra_query: Query | None = None, |
| 1059 | extra_body: Body | None = None, |
| 1060 | timeout: float | httpx.Timeout | None | NotGiven = not_given, |
| 1061 | poll_interval_ms: int | Omit = omit, |
| 1062 | ) -> Run: |
| 1063 | """ |
| 1064 | A helper to poll a run status until it reaches a terminal state. More |
| 1065 | information on Run lifecycles can be found here: |
| 1066 | https://platform.openai.com/docs/assistants/how-it-works/runs-and-run-steps |
| 1067 | """ |
| 1068 | extra_headers = {"X-Stainless-Poll-Helper": "true", **(extra_headers or {})} |
| 1069 | |
| 1070 | if is_given(poll_interval_ms): |
| 1071 | extra_headers["X-Stainless-Custom-Poll-Interval"] = str(poll_interval_ms) |
| 1072 | |
| 1073 | terminal_states = {"requires_action", "cancelled", "completed", "failed", "expired", "incomplete"} |
| 1074 | while True: |
| 1075 | response = self.with_raw_response.retrieve( # pyright: ignore[reportDeprecated] |
| 1076 | thread_id=thread_id, |
| 1077 | run_id=run_id, |
| 1078 | extra_headers=extra_headers, |
| 1079 | extra_body=extra_body, |
| 1080 | extra_query=extra_query, |
| 1081 | timeout=timeout, |
| 1082 | ) |
| 1083 | |
| 1084 | run = response.parse() |
| 1085 | # Return if we reached a terminal state |
| 1086 | if run.status in terminal_states: |
| 1087 | return run |
| 1088 | |
| 1089 | if not is_given(poll_interval_ms): |
| 1090 | from_header = response.headers.get("openai-poll-after-ms") |
| 1091 | if from_header is not None: |
| 1092 | poll_interval_ms = int(from_header) |
| 1093 | else: |
| 1094 | poll_interval_ms = 1000 |
| 1095 | |
| 1096 | self._sleep(poll_interval_ms / 1000) |
| 1097 | |
| 1098 | @overload |
| 1099 | @typing_extensions.deprecated("The Assistants API is deprecated in favor of the Responses API") |
no test coverage detected