ByoungSooPark/kittycad.py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add websockets lib, bump version (#116)

Browse Source 
* add websockets lib, bump version

Signed-off-by: Jess Frazelle <github@jessfraz.com>

* use template

Signed-off-by: Jess Frazelle <github@jessfraz.com>

* fixes

Signed-off-by: Jess Frazelle <github@jessfraz.com>

* I have generated the latest API!

---------

Signed-off-by: Jess Frazelle <github@jessfraz.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
This commit is contained in:
Jess Frazelle
2023-07-31 18:05:13 -07:00
committed by GitHub
parent 20b2a188dc
commit 1a978b79f0
77 changed files with 871 additions and 678 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
kittycad/api/executor/create_executor_term.py
View File

@ -1,9 +1,10 @@
from typing import Any, Dict
import httpx
from websockets.client import WebSocketClientProtocol, connect as ws_connect_async
from websockets.sync.client import ClientConnection, connect as ws_connect
from ...client import Client
from ...types import Response
from ...models.error import Error
def _get_kwargs(
@ -23,68 +24,40 @@ def _get_kwargs(
}
def _parse_response(*, response: httpx.Response):
return
def _build_response(*, response: httpx.Response) -> Response[Any]:
return Response(
status_code=response.status_code,
content=response.content,
headers=response.headers,
parsed=_parse_response(response=response),
)
def sync_detailed(
*,
client: Client,
) -> Response[Any]:
kwargs = _get_kwargs(
client=client,
)
response = httpx.get(
verify=client.verify_ssl,
**kwargs,
)
return _build_response(response=response)
def sync(
*,
client: Client,
):
) -> ClientConnection:
"""Attach to a docker container to create an interactive terminal.""" # noqa: E501
return sync_detailed(
client=client,
).parsed
async def asyncio_detailed(
*,
client: Client,
) -> Response[Any]:
kwargs = _get_kwargs(
client=client,
)
async with httpx.AsyncClient(verify=client.verify_ssl) as _client:
response = await _client.get(**kwargs)
with ws_connect(
kwargs["url"].replace("https://", "wss://"),
additional_headers=kwargs["headers"],
) as websocket:
return websocket # type: ignore
return _build_response(response=response)
# Return an error if we got here.
return Error(message="An error occurred while connecting to the websocket.")
async def asyncio(
*,
client: Client,
):
) -> WebSocketClientProtocol:
"""Attach to a docker container to create an interactive terminal.""" # noqa: E501
return (
await asyncio_detailed(
client=client,
)
).parsed
kwargs = _get_kwargs(
client=client,
)
async with ws_connect_async(
kwargs["url"].replace("https://", "wss://"), extra_headers=kwargs["headers"]
) as websocket:
return websocket
# Return an error if we got here.
return Error(message="An error occurred while connecting to the websocket.")