Provides the core interface to iterate over a synchronous stream response.
| 19 | |
| 20 | |
| 21 | class Stream(Generic[_T]): |
| 22 | """Provides the core interface to iterate over a synchronous stream response.""" |
| 23 | |
| 24 | response: httpx.Response |
| 25 | |
| 26 | _decoder: SSEBytesDecoder |
| 27 | |
| 28 | def __init__( |
| 29 | self, |
| 30 | *, |
| 31 | cast_to: type[_T], |
| 32 | response: httpx.Response, |
| 33 | client: Opencode, |
| 34 | ) -> None: |
| 35 | self.response = response |
| 36 | self._cast_to = cast_to |
| 37 | self._client = client |
| 38 | self._decoder = client._make_sse_decoder() |
| 39 | self._iterator = self.__stream__() |
| 40 | |
| 41 | def __next__(self) -> _T: |
| 42 | return self._iterator.__next__() |
| 43 | |
| 44 | def __iter__(self) -> Iterator[_T]: |
| 45 | for item in self._iterator: |
| 46 | yield item |
| 47 | |
| 48 | def _iter_events(self) -> Iterator[ServerSentEvent]: |
| 49 | yield from self._decoder.iter_bytes(self.response.iter_bytes()) |
| 50 | |
| 51 | def __stream__(self) -> Iterator[_T]: |
| 52 | cast_to = cast(Any, self._cast_to) |
| 53 | response = self.response |
| 54 | process_data = self._client._process_response_data |
| 55 | iterator = self._iter_events() |
| 56 | |
| 57 | for sse in iterator: |
| 58 | yield process_data(data=sse.json(), cast_to=cast_to, response=response) |
| 59 | |
| 60 | # Ensure the entire stream is consumed |
| 61 | for _sse in iterator: |
| 62 | ... |
| 63 | |
| 64 | def __enter__(self) -> Self: |
| 65 | return self |
| 66 | |
| 67 | def __exit__( |
| 68 | self, |
| 69 | exc_type: type[BaseException] | None, |
| 70 | exc: BaseException | None, |
| 71 | exc_tb: TracebackType | None, |
| 72 | ) -> None: |
| 73 | self.close() |
| 74 | |
| 75 | def close(self) -> None: |
| 76 | """ |
| 77 | Close the response and release the connection. |
| 78 |
no outgoing calls