Skip to content

Commit 1d7a848

Browse files
stainless-app[bot]stainless-app[bot]
committed
feat(api): add events streaming
1 parent ec5eb7e commit 1d7a848

File tree

6 files changed

+171
-28
lines changed

6 files changed

+171
-28
lines changed

Diff for: ‎.stats.yml

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,2 @@
11
configured_endpoints: 106
2-
openapi_spec_url: https://storage.googleapis.com/stainless-sdk-openapi-specs/gitpod%2Fgitpod-2e9f8b8666b2fd4e346a3acbf81a2c82a6f3793e01bc146499708efaf0c250c5.yml
2+
openapi_spec_url: https://storage.googleapis.com/stainless-sdk-openapi-specs/gitpod%2Fgitpod-922f204ec36b8a84ae8f96e73923e92cb2044a14c6497d173f4b7110a090ac30.yml

Diff for: ‎api.md

+1-1
Original file line numberDiff line numberDiff line change
@@ -167,7 +167,7 @@ from gitpod.types import EventListResponse, EventWatchResponse
167167
Methods:
168168

169169
- <code title="get /gitpod.v1.EventService/ListAuditLogs">client.events.<a href="./src/gitpod/resources/events.py">list</a>(\*\*<a href="src/gitpod/types/event_list_params.py">params</a>) -> <a href="./src/gitpod/types/event_list_response.py">EventListResponse</a></code>
170-
- <code title="post /gitpod.v1.EventService/WatchEvents">client.events.<a href="./src/gitpod/resources/events.py">watch</a>(\*\*<a href="src/gitpod/types/event_watch_params.py">params</a>) -> <a href="./src/gitpod/types/event_watch_response.py">EventWatchResponse</a></code>
170+
- <code title="post /gitpod.v1.EventService/WatchEvents">client.events.<a href="./src/gitpod/resources/events.py">watch</a>(\*\*<a href="src/gitpod/types/event_watch_params.py">params</a>) -> <a href="./src/gitpod/types/event_watch_response.py">JSONLDecoder[EventWatchResponse]</a></code>
171171

172172
# Groups
173173

Diff for: ‎src/gitpod/_decoders/jsonl.py

+101
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
from __future__ import annotations
2+
3+
import json
4+
from typing_extensions import Generic, TypeVar, Iterator, AsyncIterator
5+
6+
import httpx
7+
8+
from .._models import construct_type_unchecked
9+
10+
_T = TypeVar("_T")
11+
12+
13+
class JSONLDecoder(Generic[_T]):
14+
"""A decoder for [JSON Lines](https://jsonlines.org) format.
15+
16+
This class provides an iterator over a byte-iterator that parses each JSON Line
17+
into a given type.
18+
"""
19+
20+
http_response: httpx.Response | None
21+
"""The HTTP response this decoder was constructed from"""
22+
23+
def __init__(
24+
self, *, raw_iterator: Iterator[bytes], line_type: type[_T], http_response: httpx.Response | None
25+
) -> None:
26+
super().__init__()
27+
self.http_response = http_response
28+
self._raw_iterator = raw_iterator
29+
self._line_type = line_type
30+
self._iterator = self.__decode__()
31+
32+
def __decode__(self) -> Iterator[_T]:
33+
buf = b""
34+
for chunk in self._raw_iterator:
35+
for line in chunk.splitlines(keepends=True):
36+
buf += line
37+
if buf.endswith((b"\r", b"\n", b"\r\n")):
38+
yield construct_type_unchecked(
39+
value=json.loads(buf),
40+
type_=self._line_type,
41+
)
42+
buf = b""
43+
44+
# flush
45+
if buf:
46+
yield construct_type_unchecked(
47+
value=json.loads(buf),
48+
type_=self._line_type,
49+
)
50+
51+
def __next__(self) -> _T:
52+
return self._iterator.__next__()
53+
54+
def __iter__(self) -> Iterator[_T]:
55+
for item in self._iterator:
56+
yield item
57+
58+
59+
class AsyncJSONLDecoder(Generic[_T]):
60+
"""A decoder for [JSON Lines](https://jsonlines.org) format.
61+
62+
This class provides an async iterator over a byte-iterator that parses each JSON Line
63+
into a given type.
64+
"""
65+
66+
http_response: httpx.Response | None
67+
68+
def __init__(
69+
self, *, raw_iterator: AsyncIterator[bytes], line_type: type[_T], http_response: httpx.Response | None
70+
) -> None:
71+
super().__init__()
72+
self.http_response = http_response
73+
self._raw_iterator = raw_iterator
74+
self._line_type = line_type
75+
self._iterator = self.__decode__()
76+
77+
async def __decode__(self) -> AsyncIterator[_T]:
78+
buf = b""
79+
async for chunk in self._raw_iterator:
80+
for line in chunk.splitlines(keepends=True):
81+
buf += line
82+
if buf.endswith((b"\r", b"\n", b"\r\n")):
83+
yield construct_type_unchecked(
84+
value=json.loads(buf),
85+
type_=self._line_type,
86+
)
87+
buf = b""
88+
89+
# flush
90+
if buf:
91+
yield construct_type_unchecked(
92+
value=json.loads(buf),
93+
type_=self._line_type,
94+
)
95+
96+
async def __anext__(self) -> _T:
97+
return await self._iterator.__anext__()
98+
99+
async def __aiter__(self) -> AsyncIterator[_T]:
100+
async for item in self._iterator:
101+
yield item

Diff for: ‎src/gitpod/_response.py

+22
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
from ._constants import RAW_RESPONSE_HEADER, OVERRIDE_CAST_TO_HEADER
3131
from ._streaming import Stream, AsyncStream, is_stream_class_type, extract_stream_chunk_type
3232
from ._exceptions import GitpodError, APIResponseValidationError
33+
from ._decoders.jsonl import JSONLDecoder, AsyncJSONLDecoder
3334

3435
if TYPE_CHECKING:
3536
from ._models import FinalRequestOptions
@@ -138,6 +139,27 @@ def _parse(self, *, to: type[_T] | None = None) -> R | _T:
138139

139140
origin = get_origin(cast_to) or cast_to
140141

142+
if inspect.isclass(origin):
143+
if issubclass(cast(Any, origin), JSONLDecoder):
144+
return cast(
145+
R,
146+
cast("type[JSONLDecoder[Any]]", cast_to)(
147+
raw_iterator=self.http_response.iter_bytes(chunk_size=4096),
148+
line_type=extract_type_arg(cast_to, 0),
149+
http_response=self.http_response,
150+
),
151+
)
152+
153+
if issubclass(cast(Any, origin), AsyncJSONLDecoder):
154+
return cast(
155+
R,
156+
cast("type[AsyncJSONLDecoder[Any]]", cast_to)(
157+
raw_iterator=self.http_response.aiter_bytes(chunk_size=4096),
158+
line_type=extract_type_arg(cast_to, 0),
159+
http_response=self.http_response,
160+
),
161+
)
162+
141163
if self._is_sse_stream:
142164
if to:
143165
if not is_stream_class_type(to):

Diff for: ‎src/gitpod/resources/events.py

+13-10
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
async_to_streamed_response_wrapper,
2525
)
2626
from .._base_client import make_request_options
27+
from .._decoders.jsonl import JSONLDecoder, AsyncJSONLDecoder
2728
from ..types.event_list_response import EventListResponse
2829
from ..types.event_watch_response import EventWatchResponse
2930

@@ -136,7 +137,7 @@ def watch(
136137
extra_query: Query | None = None,
137138
extra_body: Body | None = None,
138139
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
139-
) -> EventWatchResponse:
140+
) -> JSONLDecoder[EventWatchResponse]:
140141
"""
141142
WatchEvents streams all requests events to the client
142143
@@ -173,7 +174,7 @@ def watch(
173174
extra_query: Query | None = None,
174175
extra_body: Body | None = None,
175176
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
176-
) -> EventWatchResponse:
177+
) -> JSONLDecoder[EventWatchResponse]:
177178
"""
178179
WatchEvents streams all requests events to the client
179180
@@ -211,8 +212,8 @@ def watch(
211212
extra_query: Query | None = None,
212213
extra_body: Body | None = None,
213214
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
214-
) -> EventWatchResponse:
215-
extra_headers = {"Accept": "application/connect+json", **(extra_headers or {})}
215+
) -> JSONLDecoder[EventWatchResponse]:
216+
extra_headers = {"Accept": "application/jsonl", **(extra_headers or {})}
216217
extra_headers = {
217218
**strip_not_given(
218219
{
@@ -234,7 +235,8 @@ def watch(
234235
options=make_request_options(
235236
extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout
236237
),
237-
cast_to=EventWatchResponse,
238+
cast_to=JSONLDecoder[EventWatchResponse],
239+
stream=True,
238240
)
239241

240242

@@ -344,7 +346,7 @@ async def watch(
344346
extra_query: Query | None = None,
345347
extra_body: Body | None = None,
346348
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
347-
) -> EventWatchResponse:
349+
) -> AsyncJSONLDecoder[EventWatchResponse]:
348350
"""
349351
WatchEvents streams all requests events to the client
350352
@@ -381,7 +383,7 @@ async def watch(
381383
extra_query: Query | None = None,
382384
extra_body: Body | None = None,
383385
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
384-
) -> EventWatchResponse:
386+
) -> AsyncJSONLDecoder[EventWatchResponse]:
385387
"""
386388
WatchEvents streams all requests events to the client
387389
@@ -419,8 +421,8 @@ async def watch(
419421
extra_query: Query | None = None,
420422
extra_body: Body | None = None,
421423
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
422-
) -> EventWatchResponse:
423-
extra_headers = {"Accept": "application/connect+json", **(extra_headers or {})}
424+
) -> AsyncJSONLDecoder[EventWatchResponse]:
425+
extra_headers = {"Accept": "application/jsonl", **(extra_headers or {})}
424426
extra_headers = {
425427
**strip_not_given(
426428
{
@@ -442,7 +444,8 @@ async def watch(
442444
options=make_request_options(
443445
extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout
444446
),
445-
cast_to=EventWatchResponse,
447+
cast_to=AsyncJSONLDecoder[EventWatchResponse],
448+
stream=True,
446449
)
447450

448451

0 commit comments

Comments
 (0)