Skip to content

Commit f2f6260

Browse files
alb-rlalb-rl
committed
Update streaming APIs to use stream_stdout_updates and stream_stderr_updates
1 parent 16bde8b commit f2f6260

2 files changed

Lines changed: 292 additions & 43 deletions

File tree

‎src/runloop_api_client/resources/devboxes/executions.py‎

Lines changed: 261 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -326,7 +326,7 @@ def kill(
326326
cast_to=DevboxAsyncExecutionDetailView,
327327
)
328328

329-
def stream_updates(
329+
def stream_stdout_updates(
330330
self,
331331
execution_id: str,
332332
*,
@@ -340,7 +340,7 @@ def stream_updates(
340340
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
341341
) -> Stream[ExecutionUpdateChunk]:
342342
"""
343-
Tails the logs for the given execution with SSE streaming
343+
Tails the stdout logs for the given execution with SSE streaming
344344
345345
Args:
346346
offset: The byte offset to start the stream from
@@ -357,10 +357,9 @@ def stream_updates(
357357
raise ValueError(f"Expected a non-empty value for `devbox_id` but received {devbox_id!r}")
358358
if not execution_id:
359359
raise ValueError(f"Expected a non-empty value for `execution_id` but received {execution_id!r}")
360-
# If caller requested a raw or streaming response wrapper, return the underlying stream as-is
361360
if extra_headers and extra_headers.get(RAW_RESPONSE_HEADER):
362361
return self._get(
363-
f"/v1/devboxes/{devbox_id}/executions/{execution_id}/stream_updates",
362+
f"/v1/devboxes/{devbox_id}/executions/{execution_id}/stream_stdout_updates",
364363
options=make_request_options(
365364
extra_headers=extra_headers,
366365
extra_query=extra_query,
@@ -375,11 +374,89 @@ def stream_updates(
375374
stream_cls=Stream[ExecutionUpdateChunk],
376375
)
377376

378-
# Otherwise, wrap with auto-reconnect using last seen offset
379377
def create_stream(last_offset: str | None) -> Stream[ExecutionUpdateChunk]:
380378
new_offset = last_offset if last_offset is not None else (None if isinstance(offset, NotGiven) else offset)
381379
return self._get(
382-
f"/v1/devboxes/{devbox_id}/executions/{execution_id}/stream_updates",
380+
f"/v1/devboxes/{devbox_id}/executions/{execution_id}/stream_stdout_updates",
381+
options=make_request_options(
382+
extra_headers=extra_headers,
383+
extra_query=extra_query,
384+
extra_body=extra_body,
385+
timeout=timeout,
386+
query=maybe_transform(
387+
{"offset": new_offset}, execution_stream_updates_params.ExecutionStreamUpdatesParams
388+
),
389+
),
390+
cast_to=DevboxAsyncExecutionDetailView,
391+
stream=True,
392+
stream_cls=Stream[ExecutionUpdateChunk],
393+
)
394+
395+
initial_stream = create_stream(None)
396+
397+
def get_offset(item: ExecutionUpdateChunk) -> str | None:
398+
value = getattr(item, "offset", None)
399+
if value is None:
400+
return None
401+
return str(value)
402+
403+
return cast(
404+
Stream[ExecutionUpdateChunk],
405+
ReconnectingStream(current_stream=initial_stream, stream_creator=create_stream, get_offset=get_offset),
406+
)
407+
408+
def stream_stderr_updates(
409+
self,
410+
execution_id: str,
411+
*,
412+
devbox_id: str,
413+
offset: str | NotGiven = NOT_GIVEN,
414+
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
415+
# The extra values given here take precedence over values defined on the client or passed to this method.
416+
extra_headers: Headers | None = None,
417+
extra_query: Query | None = None,
418+
extra_body: Body | None = None,
419+
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
420+
) -> Stream[ExecutionUpdateChunk]:
421+
"""
422+
Tails the stderr logs for the given execution with SSE streaming
423+
424+
Args:
425+
offset: The byte offset to start the stream from
426+
427+
extra_headers: Send extra headers
428+
429+
extra_query: Add additional query parameters to the request
430+
431+
extra_body: Add additional JSON properties to the request
432+
433+
timeout: Override the client-level default timeout for this request, in seconds
434+
"""
435+
if not devbox_id:
436+
raise ValueError(f"Expected a non-empty value for `devbox_id` but received {devbox_id!r}")
437+
if not execution_id:
438+
raise ValueError(f"Expected a non-empty value for `execution_id` but received {execution_id!r}")
439+
if extra_headers and extra_headers.get(RAW_RESPONSE_HEADER):
440+
return self._get(
441+
f"/v1/devboxes/{devbox_id}/executions/{execution_id}/stream_stderr_updates",
442+
options=make_request_options(
443+
extra_headers=extra_headers,
444+
extra_query=extra_query,
445+
extra_body=extra_body,
446+
timeout=timeout,
447+
query=maybe_transform(
448+
{"offset": offset}, execution_stream_updates_params.ExecutionStreamUpdatesParams
449+
),
450+
),
451+
cast_to=DevboxAsyncExecutionDetailView,
452+
stream=True,
453+
stream_cls=Stream[ExecutionUpdateChunk],
454+
)
455+
456+
def create_stream(last_offset: str | None) -> Stream[ExecutionUpdateChunk]:
457+
new_offset = last_offset if last_offset is not None else (None if isinstance(offset, NotGiven) else offset)
458+
return self._get(
459+
f"/v1/devboxes/{devbox_id}/executions/{execution_id}/stream_stderr_updates",
383460
options=make_request_options(
384461
extra_headers=extra_headers,
385462
extra_query=extra_query,
@@ -768,6 +845,164 @@ def get_offset(item: ExecutionUpdateChunk) -> str | None:
768845
AsyncReconnectingStream(current_stream=initial_stream, stream_creator=create_stream, get_offset=get_offset),
769846
)
770847

848+
async def stream_stdout_updates(
849+
self,
850+
execution_id: str,
851+
*,
852+
devbox_id: str,
853+
offset: str | NotGiven = NOT_GIVEN,
854+
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
855+
# The extra values given here take precedence over values defined on the client or passed to this method.
856+
extra_headers: Headers | None = None,
857+
extra_query: Query | None = None,
858+
extra_body: Body | None = None,
859+
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
860+
) -> AsyncStream[ExecutionUpdateChunk]:
861+
"""
862+
Tails the stdout logs for the given execution with SSE streaming
863+
864+
Args:
865+
offset: The byte offset to start the stream from
866+
867+
extra_headers: Send extra headers
868+
869+
extra_query: Add additional query parameters to the request
870+
871+
extra_body: Add additional JSON properties to the request
872+
873+
timeout: Override the client-level default timeout for this request, in seconds
874+
"""
875+
if not devbox_id:
876+
raise ValueError(f"Expected a non-empty value for `devbox_id` but received {devbox_id!r}")
877+
if not execution_id:
878+
raise ValueError(f"Expected a non-empty value for `execution_id` but received {execution_id!r}")
879+
if extra_headers and extra_headers.get(RAW_RESPONSE_HEADER):
880+
return await self._get(
881+
f"/v1/devboxes/{devbox_id}/executions/{execution_id}/stream_stdout_updates",
882+
options=make_request_options(
883+
extra_headers=extra_headers,
884+
extra_query=extra_query,
885+
extra_body=extra_body,
886+
timeout=timeout,
887+
query=await async_maybe_transform(
888+
{"offset": offset}, execution_stream_updates_params.ExecutionStreamUpdatesParams
889+
),
890+
),
891+
cast_to=DevboxAsyncExecutionDetailView,
892+
stream=True,
893+
stream_cls=AsyncStream[ExecutionUpdateChunk],
894+
)
895+
896+
async def create_stream(last_offset: str | None) -> AsyncStream[ExecutionUpdateChunk]:
897+
new_offset = last_offset if last_offset is not None else (None if isinstance(offset, NotGiven) else offset)
898+
return await self._get(
899+
f"/v1/devboxes/{devbox_id}/executions/{execution_id}/stream_stdout_updates",
900+
options=make_request_options(
901+
extra_headers=extra_headers,
902+
extra_query=extra_query,
903+
extra_body=extra_body,
904+
timeout=timeout,
905+
query=await async_maybe_transform(
906+
{"offset": new_offset}, execution_stream_updates_params.ExecutionStreamUpdatesParams
907+
),
908+
),
909+
cast_to=DevboxAsyncExecutionDetailView,
910+
stream=True,
911+
stream_cls=AsyncStream[ExecutionUpdateChunk],
912+
)
913+
914+
initial_stream = await create_stream(None)
915+
916+
def get_offset(item: ExecutionUpdateChunk) -> str | None:
917+
value = getattr(item, "offset", None)
918+
if value is None:
919+
return None
920+
return str(value)
921+
922+
return cast(
923+
AsyncStream[ExecutionUpdateChunk],
924+
AsyncReconnectingStream(current_stream=initial_stream, stream_creator=create_stream, get_offset=get_offset),
925+
)
926+
927+
async def stream_stderr_updates(
928+
self,
929+
execution_id: str,
930+
*,
931+
devbox_id: str,
932+
offset: str | NotGiven = NOT_GIVEN,
933+
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
934+
# The extra values given here take precedence over values defined on the client or passed to this method.
935+
extra_headers: Headers | None = None,
936+
extra_query: Query | None = None,
937+
extra_body: Body | None = None,
938+
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
939+
) -> AsyncStream[ExecutionUpdateChunk]:
940+
"""
941+
Tails the stderr logs for the given execution with SSE streaming
942+
943+
Args:
944+
offset: The byte offset to start the stream from
945+
946+
extra_headers: Send extra headers
947+
948+
extra_query: Add additional query parameters to the request
949+
950+
extra_body: Add additional JSON properties to the request
951+
952+
timeout: Override the client-level default timeout for this request, in seconds
953+
"""
954+
if not devbox_id:
955+
raise ValueError(f"Expected a non-empty value for `devbox_id` but received {devbox_id!r}")
956+
if not execution_id:
957+
raise ValueError(f"Expected a non-empty value for `execution_id` but received {execution_id!r}")
958+
if extra_headers and extra_headers.get(RAW_RESPONSE_HEADER):
959+
return await self._get(
960+
f"/v1/devboxes/{devbox_id}/executions/{execution_id}/stream_stderr_updates",
961+
options=make_request_options(
962+
extra_headers=extra_headers,
963+
extra_query=extra_query,
964+
extra_body=extra_body,
965+
timeout=timeout,
966+
query=await async_maybe_transform(
967+
{"offset": offset}, execution_stream_updates_params.ExecutionStreamUpdatesParams
968+
),
969+
),
970+
cast_to=DevboxAsyncExecutionDetailView,
971+
stream=True,
972+
stream_cls=AsyncStream[ExecutionUpdateChunk],
973+
)
974+
975+
async def create_stream(last_offset: str | None) -> AsyncStream[ExecutionUpdateChunk]:
976+
new_offset = last_offset if last_offset is not None else (None if isinstance(offset, NotGiven) else offset)
977+
return await self._get(
978+
f"/v1/devboxes/{devbox_id}/executions/{execution_id}/stream_stderr_updates",
979+
options=make_request_options(
980+
extra_headers=extra_headers,
981+
extra_query=extra_query,
982+
extra_body=extra_body,
983+
timeout=timeout,
984+
query=await async_maybe_transform(
985+
{"offset": new_offset}, execution_stream_updates_params.ExecutionStreamUpdatesParams
986+
),
987+
),
988+
cast_to=DevboxAsyncExecutionDetailView,
989+
stream=True,
990+
stream_cls=AsyncStream[ExecutionUpdateChunk],
991+
)
992+
993+
initial_stream = await create_stream(None)
994+
995+
def get_offset(item: ExecutionUpdateChunk) -> str | None:
996+
value = getattr(item, "offset", None)
997+
if value is None:
998+
return None
999+
return str(value)
1000+
1001+
return cast(
1002+
AsyncStream[ExecutionUpdateChunk],
1003+
AsyncReconnectingStream(current_stream=initial_stream, stream_creator=create_stream, get_offset=get_offset),
1004+
)
1005+
7711006

7721007
class ExecutionsResourceWithRawResponse:
7731008
def __init__(self, executions: ExecutionsResource) -> None:
@@ -785,8 +1020,11 @@ def __init__(self, executions: ExecutionsResource) -> None:
7851020
self.kill = to_raw_response_wrapper(
7861021
executions.kill,
7871022
)
788-
self.stream_updates = to_raw_response_wrapper(
789-
executions.stream_updates,
1023+
self.stream_stdout_updates = to_raw_response_wrapper(
1024+
executions.stream_stdout_updates,
1025+
)
1026+
self.stream_stderr_updates = to_raw_response_wrapper(
1027+
executions.stream_stderr_updates,
7901028
)
7911029

7921030

@@ -806,8 +1044,11 @@ def __init__(self, executions: AsyncExecutionsResource) -> None:
8061044
self.kill = async_to_raw_response_wrapper(
8071045
executions.kill,
8081046
)
809-
self.stream_updates = async_to_raw_response_wrapper(
810-
executions.stream_updates,
1047+
self.stream_stdout_updates = async_to_raw_response_wrapper(
1048+
executions.stream_stdout_updates,
1049+
)
1050+
self.stream_stderr_updates = async_to_raw_response_wrapper(
1051+
executions.stream_stderr_updates,
8111052
)
8121053

8131054

@@ -827,8 +1068,11 @@ def __init__(self, executions: ExecutionsResource) -> None:
8271068
self.kill = to_streamed_response_wrapper(
8281069
executions.kill,
8291070
)
830-
self.stream_updates = to_streamed_response_wrapper(
831-
executions.stream_updates,
1071+
self.stream_stdout_updates = to_streamed_response_wrapper(
1072+
executions.stream_stdout_updates,
1073+
)
1074+
self.stream_stderr_updates = to_streamed_response_wrapper(
1075+
executions.stream_stderr_updates,
8321076
)
8331077

8341078

@@ -848,6 +1092,9 @@ def __init__(self, executions: AsyncExecutionsResource) -> None:
8481092
self.kill = async_to_streamed_response_wrapper(
8491093
executions.kill,
8501094
)
851-
self.stream_updates = async_to_streamed_response_wrapper(
852-
executions.stream_updates,
1095+
self.stream_stdout_updates = async_to_streamed_response_wrapper(
1096+
executions.stream_stdout_updates,
1097+
)
1098+
self.stream_stderr_updates = async_to_streamed_response_wrapper(
1099+
executions.stream_stderr_updates,
8531100
)

0 commit comments

Comments
 (0)