Skip to content

Commit 6e50de6

Browse files
Leonid Ryzhykryzhyk
Leonid Ryzhyk
authored and
ryzhyk
committed
Export JSON update format spec via OpenAPI.
Signed-off-by: Leonid Ryzhyk <leonid@feldera.com>
1 parent 581a499 commit 6e50de6

4 files changed

Lines changed: 78 additions & 2 deletions

File tree

‎crates/pipeline_manager/src/api.rs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1742,6 +1742,8 @@ pub struct PipelineIdOrNameQuery {
17421742
("pipeline_id" = Uuid, Path, description = "Unique pipeline identifier."),
17431743
("table_name" = String, Path, description = "SQL table name."),
17441744
("format" = String, Query, description = "Input data format, e.g., 'csv' or 'json'."),
1745+
("array" = Option<bool>, Query, description = "Set to `true` if updates in this stream are packaged into JSON arrays (used in conjunction with `format=json`). The default values is `false`."),
1746+
("update_format" = Option<JsonUpdateFormat>, Query, description = "JSON data change event format (used in conjunction with `format=json`). The default value is 'insert_delete'."),
17451747
),
17461748
tag = "Pipelines",
17471749
request_body(
@@ -1825,6 +1827,7 @@ async fn http_input(
18251827
("query" = Option<OutputQuery>, Query, description = "Query to execute on the table. Must be one of 'table', 'neighborhood', or 'quantiles'. The default value is 'table'"),
18261828
("mode" = Option<EgressMode>, Query, description = "Output mode. Must be one of 'watch' or 'snapshot'. The default value is 'watch'"),
18271829
("quantiles" = Option<u32>, Query, description = "For 'quantiles' queries: the number of quantiles to output. The default value is 100."),
1830+
("array" = Option<bool>, Query, description = "Set to `true` to group updates in this stream into JSON arrays (used in conjunction with `format=json`). The default value is `false`"),
18281831
),
18291832
request_body(
18301833
content = Option<NeighborhoodQuery>,

‎crates/pipeline_manager/src/integration_test.rs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -684,7 +684,10 @@ async fn json_ingress() {
684684
// Push some data using default json config.
685685
let req = config
686686
.post_json(
687-
format!("/v0/pipelines/{}/ingress/T1?format=json&update_format=raw", id),
687+
format!(
688+
"/v0/pipelines/{}/ingress/T1?format=json&update_format=raw",
689+
id
690+
),
688691
r#"{"C1": 10, "C2": true}
689692
{"C1": 20, "C3": "foo"}"#
690693
.to_string(),

‎python/feldera-api-client/feldera_api_client/api/pipelines/http_input.py‎

Lines changed: 56 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,20 +6,31 @@
66
from ... import errors
77
from ...client import AuthenticatedClient, Client
88
from ...models.error_response import ErrorResponse
9-
from ...types import UNSET, Response
9+
from ...models.json_update_format import JsonUpdateFormat
10+
from ...types import UNSET, Response, Unset
1011

1112

1213
def _get_kwargs(
1314
pipeline_id: str,
1415
table_name: str,
1516
*,
1617
format_: str,
18+
array: Union[Unset, None, bool] = UNSET,
19+
update_format: Union[Unset, None, JsonUpdateFormat] = UNSET,
1720
) -> Dict[str, Any]:
1821
pass
1922

2023
params: Dict[str, Any] = {}
2124
params["format"] = format_
2225

26+
params["array"] = array
27+
28+
json_update_format: Union[Unset, None, str] = UNSET
29+
if not isinstance(update_format, Unset):
30+
json_update_format = update_format.value if update_format else None
31+
32+
params["update_format"] = json_update_format
33+
2334
params = {k: v for k, v in params.items() if v is not UNSET and v is not None}
2435

2536
return {
@@ -73,6 +84,8 @@ def sync_detailed(
7384
*,
7485
client: Union[AuthenticatedClient, Client],
7586
format_: str,
87+
array: Union[Unset, None, bool] = UNSET,
88+
update_format: Union[Unset, None, JsonUpdateFormat] = UNSET,
7689
) -> Response[Union[Any, ErrorResponse]]:
7790
"""Push data to a SQL table.
7891
@@ -90,6 +103,13 @@ def sync_detailed(
90103
pipeline_id (str):
91104
table_name (str):
92105
format_ (str):
106+
array (Union[Unset, None, bool]):
107+
update_format (Union[Unset, None, JsonUpdateFormat]): Supported JSON data change event
108+
formats.
109+
110+
Each element in a JSON-formatted input stream specifies
111+
an update to one or more records in an input table. We support
112+
several different ways to represent such updates.
93113
94114
Raises:
95115
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
@@ -103,6 +123,8 @@ def sync_detailed(
103123
pipeline_id=pipeline_id,
104124
table_name=table_name,
105125
format_=format_,
126+
array=array,
127+
update_format=update_format,
106128
)
107129

108130
response = client.get_httpx_client().request(
@@ -118,6 +140,8 @@ def sync(
118140
*,
119141
client: Union[AuthenticatedClient, Client],
120142
format_: str,
143+
array: Union[Unset, None, bool] = UNSET,
144+
update_format: Union[Unset, None, JsonUpdateFormat] = UNSET,
121145
) -> Optional[Union[Any, ErrorResponse]]:
122146
"""Push data to a SQL table.
123147
@@ -135,6 +159,13 @@ def sync(
135159
pipeline_id (str):
136160
table_name (str):
137161
format_ (str):
162+
array (Union[Unset, None, bool]):
163+
update_format (Union[Unset, None, JsonUpdateFormat]): Supported JSON data change event
164+
formats.
165+
166+
Each element in a JSON-formatted input stream specifies
167+
an update to one or more records in an input table. We support
168+
several different ways to represent such updates.
138169
139170
Raises:
140171
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
@@ -149,6 +180,8 @@ def sync(
149180
table_name=table_name,
150181
client=client,
151182
format_=format_,
183+
array=array,
184+
update_format=update_format,
152185
).parsed
153186

154187

@@ -158,6 +191,8 @@ async def asyncio_detailed(
158191
*,
159192
client: Union[AuthenticatedClient, Client],
160193
format_: str,
194+
array: Union[Unset, None, bool] = UNSET,
195+
update_format: Union[Unset, None, JsonUpdateFormat] = UNSET,
161196
) -> Response[Union[Any, ErrorResponse]]:
162197
"""Push data to a SQL table.
163198
@@ -175,6 +210,13 @@ async def asyncio_detailed(
175210
pipeline_id (str):
176211
table_name (str):
177212
format_ (str):
213+
array (Union[Unset, None, bool]):
214+
update_format (Union[Unset, None, JsonUpdateFormat]): Supported JSON data change event
215+
formats.
216+
217+
Each element in a JSON-formatted input stream specifies
218+
an update to one or more records in an input table. We support
219+
several different ways to represent such updates.
178220
179221
Raises:
180222
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
@@ -188,6 +230,8 @@ async def asyncio_detailed(
188230
pipeline_id=pipeline_id,
189231
table_name=table_name,
190232
format_=format_,
233+
array=array,
234+
update_format=update_format,
191235
)
192236

193237
response = await client.get_async_httpx_client().request(**kwargs)
@@ -201,6 +245,8 @@ async def asyncio(
201245
*,
202246
client: Union[AuthenticatedClient, Client],
203247
format_: str,
248+
array: Union[Unset, None, bool] = UNSET,
249+
update_format: Union[Unset, None, JsonUpdateFormat] = UNSET,
204250
) -> Optional[Union[Any, ErrorResponse]]:
205251
"""Push data to a SQL table.
206252
@@ -218,6 +264,13 @@ async def asyncio(
218264
pipeline_id (str):
219265
table_name (str):
220266
format_ (str):
267+
array (Union[Unset, None, bool]):
268+
update_format (Union[Unset, None, JsonUpdateFormat]): Supported JSON data change event
269+
formats.
270+
271+
Each element in a JSON-formatted input stream specifies
272+
an update to one or more records in an input table. We support
273+
several different ways to represent such updates.
221274
222275
Raises:
223276
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
@@ -233,5 +286,7 @@ async def asyncio(
233286
table_name=table_name,
234287
client=client,
235288
format_=format_,
289+
array=array,
290+
update_format=update_format,
236291
)
237292
).parsed

‎python/feldera-api-client/feldera_api_client/api/pipelines/http_output.py‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ def _get_kwargs(
2222
query: Union[Unset, None, OutputQuery] = UNSET,
2323
mode: Union[Unset, None, EgressMode] = UNSET,
2424
quantiles: Union[Unset, None, int] = UNSET,
25+
array: Union[Unset, None, bool] = UNSET,
2526
) -> Dict[str, Any]:
2627
pass
2728

@@ -42,6 +43,8 @@ def _get_kwargs(
4243

4344
params["quantiles"] = quantiles
4445

46+
params["array"] = array
47+
4548
params = {k: v for k, v in params.items() if v is not UNSET and v is not None}
4649

4750
json_json_body = json_body.to_dict() if json_body else None
@@ -107,6 +110,7 @@ def sync_detailed(
107110
query: Union[Unset, None, OutputQuery] = UNSET,
108111
mode: Union[Unset, None, EgressMode] = UNSET,
109112
quantiles: Union[Unset, None, int] = UNSET,
113+
array: Union[Unset, None, bool] = UNSET,
110114
) -> Response[Union[Chunk, ErrorResponse]]:
111115
"""Subscribe to a stream of updates from a SQL view or table.
112116
@@ -129,6 +133,7 @@ def sync_detailed(
129133
three pre-defined queries to inspect the contents of a table or view.
130134
mode (Union[Unset, None, EgressMode]):
131135
quantiles (Union[Unset, None, int]):
136+
array (Union[Unset, None, bool]):
132137
json_body (Optional[NeighborhoodQuery]): A request to output a specific neighborhood of a
133138
table or view.
134139
The neighborhood is defined in terms of its central point (`anchor`)
@@ -150,6 +155,7 @@ def sync_detailed(
150155
query=query,
151156
mode=mode,
152157
quantiles=quantiles,
158+
array=array,
153159
)
154160

155161
response = client.get_httpx_client().request(
@@ -169,6 +175,7 @@ def sync(
169175
query: Union[Unset, None, OutputQuery] = UNSET,
170176
mode: Union[Unset, None, EgressMode] = UNSET,
171177
quantiles: Union[Unset, None, int] = UNSET,
178+
array: Union[Unset, None, bool] = UNSET,
172179
) -> Optional[Union[Chunk, ErrorResponse]]:
173180
"""Subscribe to a stream of updates from a SQL view or table.
174181
@@ -191,6 +198,7 @@ def sync(
191198
three pre-defined queries to inspect the contents of a table or view.
192199
mode (Union[Unset, None, EgressMode]):
193200
quantiles (Union[Unset, None, int]):
201+
array (Union[Unset, None, bool]):
194202
json_body (Optional[NeighborhoodQuery]): A request to output a specific neighborhood of a
195203
table or view.
196204
The neighborhood is defined in terms of its central point (`anchor`)
@@ -213,6 +221,7 @@ def sync(
213221
query=query,
214222
mode=mode,
215223
quantiles=quantiles,
224+
array=array,
216225
).parsed
217226

218227

@@ -226,6 +235,7 @@ async def asyncio_detailed(
226235
query: Union[Unset, None, OutputQuery] = UNSET,
227236
mode: Union[Unset, None, EgressMode] = UNSET,
228237
quantiles: Union[Unset, None, int] = UNSET,
238+
array: Union[Unset, None, bool] = UNSET,
229239
) -> Response[Union[Chunk, ErrorResponse]]:
230240
"""Subscribe to a stream of updates from a SQL view or table.
231241
@@ -248,6 +258,7 @@ async def asyncio_detailed(
248258
three pre-defined queries to inspect the contents of a table or view.
249259
mode (Union[Unset, None, EgressMode]):
250260
quantiles (Union[Unset, None, int]):
261+
array (Union[Unset, None, bool]):
251262
json_body (Optional[NeighborhoodQuery]): A request to output a specific neighborhood of a
252263
table or view.
253264
The neighborhood is defined in terms of its central point (`anchor`)
@@ -269,6 +280,7 @@ async def asyncio_detailed(
269280
query=query,
270281
mode=mode,
271282
quantiles=quantiles,
283+
array=array,
272284
)
273285

274286
response = await client.get_async_httpx_client().request(**kwargs)
@@ -286,6 +298,7 @@ async def asyncio(
286298
query: Union[Unset, None, OutputQuery] = UNSET,
287299
mode: Union[Unset, None, EgressMode] = UNSET,
288300
quantiles: Union[Unset, None, int] = UNSET,
301+
array: Union[Unset, None, bool] = UNSET,
289302
) -> Optional[Union[Chunk, ErrorResponse]]:
290303
"""Subscribe to a stream of updates from a SQL view or table.
291304
@@ -308,6 +321,7 @@ async def asyncio(
308321
three pre-defined queries to inspect the contents of a table or view.
309322
mode (Union[Unset, None, EgressMode]):
310323
quantiles (Union[Unset, None, int]):
324+
array (Union[Unset, None, bool]):
311325
json_body (Optional[NeighborhoodQuery]): A request to output a specific neighborhood of a
312326
table or view.
313327
The neighborhood is defined in terms of its central point (`anchor`)
@@ -331,5 +345,6 @@ async def asyncio(
331345
query=query,
332346
mode=mode,
333347
quantiles=quantiles,
348+
array=array,
334349
)
335350
).parsed

0 commit comments

Comments
 (0)