feat(python): add tabby-python-client (#268)

* feat(python): add tabby-python-client

* switch to setup.py
sweep/improve-logging-information
Meng Zhang 2023-06-24 09:43:16 -07:00 committed by GitHub
parent 848aff7d1f
commit 7abca766ca
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
23 changed files with 1171 additions and 0 deletions

23
clients/tabby-python-client/.gitignore vendored Normal file
View File

@ -0,0 +1,23 @@
__pycache__/
build/
dist/
*.egg-info/
.pytest_cache/
# pyenv
.python-version
# Environments
.env
.venv
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# JetBrains
.idea/
/coverage.xml
/.coverage

View File

@ -0,0 +1,89 @@
# tabby-python-client
A client library for accessing Tabby Server
## Usage
First, create a client:
```python
from tabby_client import Client
client = Client(base_url="https://api.example.com")
```
If the endpoints you're going to hit require authentication, use `AuthenticatedClient` instead:
```python
from tabby_client import AuthenticatedClient
client = AuthenticatedClient(base_url="https://api.example.com", token="SuperSecretToken")
```
Now call your endpoint and use your models:
```python
from tabby_client.models import MyDataModel
from tabby_client.api.my_tag import get_my_data_model
from tabby_client.types import Response
my_data: MyDataModel = get_my_data_model.sync(client=client)
# or if you need more info (e.g. status_code)
response: Response[MyDataModel] = get_my_data_model.sync_detailed(client=client)
```
Or do the same thing with an async version:
```python
from tabby_client.models import MyDataModel
from tabby_client.api.my_tag import get_my_data_model
from tabby_client.types import Response
my_data: MyDataModel = await get_my_data_model.asyncio(client=client)
response: Response[MyDataModel] = await get_my_data_model.asyncio_detailed(client=client)
```
By default, when you're calling an HTTPS API it will attempt to verify that SSL is working correctly. Using certificate verification is highly recommended most of the time, but sometimes you may need to authenticate to a server (especially an internal server) using a custom certificate bundle.
```python
client = AuthenticatedClient(
base_url="https://internal_api.example.com",
token="SuperSecretToken",
verify_ssl="/path/to/certificate_bundle.pem",
)
```
You can also disable certificate validation altogether, but beware that **this is a security risk**.
```python
client = AuthenticatedClient(
base_url="https://internal_api.example.com",
token="SuperSecretToken",
verify_ssl=False
)
```
There are more settings on the generated `Client` class which let you control more runtime behavior, check out the docstring on that class for more info.
Things to know:
1. Every path/method combo becomes a Python module with four functions:
1. `sync`: Blocking request that returns parsed data (if successful) or `None`
1. `sync_detailed`: Blocking request that always returns a `Request`, optionally with `parsed` set if the request was successful.
1. `asyncio`: Like `sync` but async instead of blocking
1. `asyncio_detailed`: Like `sync_detailed` but async instead of blocking
1. All path/query params, and bodies become method arguments.
1. If your endpoint had any tags on it, the first tag will be used as a module name for the function (my_tag above)
1. Any endpoint which did not have a tag will be in `tabby_client.api.default`
## Building / publishing this Client
This project uses [Poetry](https://python-poetry.org/) to manage dependencies and packaging. Here are the basics:
1. Update the metadata in pyproject.toml (e.g. authors, version)
1. If you're using a private repository, configure it with Poetry
1. `poetry config repositories.<your-repository-name> <url-to-your-repository>`
1. `poetry config http-basic.<your-repository-name> <username> <password>`
1. Publish the client with `poetry publish --build -r <your-repository-name>` or, if for public PyPI, just `poetry publish --build`
If you want to install this client into another project without publishing it (e.g. for development) then:
1. If that project **is using Poetry**, you can simply do `poetry add <path-to-this-client>` from that project
1. If that project is not using Poetry:
1. Build a wheel with `poetry build -f wheel`
1. Install that wheel from the other project `pip install <path-to-wheel>`

View File

@ -0,0 +1,16 @@
[tool.black]
line-length = 120
target_version = ['py38', 'py39', 'py310', 'py311']
exclude = '''
(
/(
| \.git
| \.venv
| \.mypy_cache
)/
)
'''
[tool.isort]
line_length = 120
profile = "black"

View File

@ -0,0 +1,18 @@
import pathlib
from setuptools import find_packages, setup
here = pathlib.Path(__file__).parent.resolve()
long_description = (here / "README.md").read_text(encoding="utf-8")
setup(
name="tabby-python-client",
version="0.1.0",
description="A client library for accessing Tabby Server",
long_description=long_description,
long_description_content_type="text/markdown",
packages=find_packages(),
python_requires=">=3.8, <4",
install_requires=["httpx >= 0.15.0, < 0.25.0", "attrs >= 21.3.0", "python-dateutil >= 2.8.0, < 3"],
package_data={"tabby_client": ["py.typed"]},
)

View File

@ -0,0 +1,7 @@
""" A client library for accessing Tabby Server """
from .client import AuthenticatedClient, Client
__all__ = (
"AuthenticatedClient",
"Client",
)

View File

@ -0,0 +1 @@
""" Contains methods for accessing the API """

View File

@ -0,0 +1,166 @@
from http import HTTPStatus
from typing import Any, Dict, Optional, Union, cast
import httpx
from ... import errors
from ...client import Client
from ...models.completion_request import CompletionRequest
from ...models.completion_response import CompletionResponse
from ...types import Response
def _get_kwargs(
*,
client: Client,
json_body: CompletionRequest,
) -> Dict[str, Any]:
url = "{}/v1/completions".format(client.base_url)
headers: Dict[str, str] = client.get_headers()
cookies: Dict[str, Any] = client.get_cookies()
json_json_body = json_body.to_dict()
return {
"method": "post",
"url": url,
"headers": headers,
"cookies": cookies,
"timeout": client.get_timeout(),
"follow_redirects": client.follow_redirects,
"json": json_json_body,
}
def _parse_response(*, client: Client, response: httpx.Response) -> Optional[Union[Any, CompletionResponse]]:
if response.status_code == HTTPStatus.OK:
response_200 = CompletionResponse.from_dict(response.json())
return response_200
if response.status_code == HTTPStatus.BAD_REQUEST:
response_400 = cast(Any, None)
return response_400
if client.raise_on_unexpected_status:
raise errors.UnexpectedStatus(response.status_code, response.content)
else:
return None
def _build_response(*, client: Client, response: httpx.Response) -> Response[Union[Any, CompletionResponse]]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
*,
client: Client,
json_body: CompletionRequest,
) -> Response[Union[Any, CompletionResponse]]:
r"""
Args:
json_body (CompletionRequest): Example: {'language': 'python', 'segments': {'prefix':
'def fib(n):\n ', 'suffix': '\n return fib(n - 1) + fib(n - 2)'}}.
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[Union[Any, CompletionResponse]]
"""
kwargs = _get_kwargs(
client=client,
json_body=json_body,
)
response = httpx.request(
verify=client.verify_ssl,
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
*,
client: Client,
json_body: CompletionRequest,
) -> Optional[Union[Any, CompletionResponse]]:
r"""
Args:
json_body (CompletionRequest): Example: {'language': 'python', 'segments': {'prefix':
'def fib(n):\n ', 'suffix': '\n return fib(n - 1) + fib(n - 2)'}}.
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Union[Any, CompletionResponse]
"""
return sync_detailed(
client=client,
json_body=json_body,
).parsed
async def asyncio_detailed(
*,
client: Client,
json_body: CompletionRequest,
) -> Response[Union[Any, CompletionResponse]]:
r"""
Args:
json_body (CompletionRequest): Example: {'language': 'python', 'segments': {'prefix':
'def fib(n):\n ', 'suffix': '\n return fib(n - 1) + fib(n - 2)'}}.
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[Union[Any, CompletionResponse]]
"""
kwargs = _get_kwargs(
client=client,
json_body=json_body,
)
async with httpx.AsyncClient(verify=client.verify_ssl) as _client:
response = await _client.request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
*,
client: Client,
json_body: CompletionRequest,
) -> Optional[Union[Any, CompletionResponse]]:
r"""
Args:
json_body (CompletionRequest): Example: {'language': 'python', 'segments': {'prefix':
'def fib(n):\n ', 'suffix': '\n return fib(n - 1) + fib(n - 2)'}}.
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Union[Any, CompletionResponse]
"""
return (
await asyncio_detailed(
client=client,
json_body=json_body,
)
).parsed

View File

@ -0,0 +1,110 @@
from http import HTTPStatus
from typing import Any, Dict, Optional
import httpx
from ... import errors
from ...client import Client
from ...models.log_event_request import LogEventRequest
from ...types import Response
def _get_kwargs(
*,
client: Client,
json_body: LogEventRequest,
) -> Dict[str, Any]:
url = "{}/v1/events".format(client.base_url)
headers: Dict[str, str] = client.get_headers()
cookies: Dict[str, Any] = client.get_cookies()
json_json_body = json_body.to_dict()
return {
"method": "post",
"url": url,
"headers": headers,
"cookies": cookies,
"timeout": client.get_timeout(),
"follow_redirects": client.follow_redirects,
"json": json_json_body,
}
def _parse_response(*, client: Client, response: httpx.Response) -> Optional[Any]:
if response.status_code == HTTPStatus.OK:
return None
if response.status_code == HTTPStatus.BAD_REQUEST:
return None
if client.raise_on_unexpected_status:
raise errors.UnexpectedStatus(response.status_code, response.content)
else:
return None
def _build_response(*, client: Client, response: httpx.Response) -> Response[Any]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
*,
client: Client,
json_body: LogEventRequest,
) -> Response[Any]:
"""
Args:
json_body (LogEventRequest):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[Any]
"""
kwargs = _get_kwargs(
client=client,
json_body=json_body,
)
response = httpx.request(
verify=client.verify_ssl,
**kwargs,
)
return _build_response(client=client, response=response)
async def asyncio_detailed(
*,
client: Client,
json_body: LogEventRequest,
) -> Response[Any]:
"""
Args:
json_body (LogEventRequest):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[Any]
"""
kwargs = _get_kwargs(
client=client,
json_body=json_body,
)
async with httpx.AsyncClient(verify=client.verify_ssl) as _client:
response = await _client.request(**kwargs)
return _build_response(client=client, response=response)

View File

@ -0,0 +1,134 @@
from http import HTTPStatus
from typing import Any, Dict, Optional
import httpx
from ... import errors
from ...client import Client
from ...models.health_state import HealthState
from ...types import Response
def _get_kwargs(
*,
client: Client,
) -> Dict[str, Any]:
url = "{}/v1/health".format(client.base_url)
headers: Dict[str, str] = client.get_headers()
cookies: Dict[str, Any] = client.get_cookies()
return {
"method": "post",
"url": url,
"headers": headers,
"cookies": cookies,
"timeout": client.get_timeout(),
"follow_redirects": client.follow_redirects,
}
def _parse_response(*, client: Client, response: httpx.Response) -> Optional[HealthState]:
if response.status_code == HTTPStatus.OK:
response_200 = HealthState.from_dict(response.json())
return response_200
if client.raise_on_unexpected_status:
raise errors.UnexpectedStatus(response.status_code, response.content)
else:
return None
def _build_response(*, client: Client, response: httpx.Response) -> Response[HealthState]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
*,
client: Client,
) -> Response[HealthState]:
"""
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[HealthState]
"""
kwargs = _get_kwargs(
client=client,
)
response = httpx.request(
verify=client.verify_ssl,
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
*,
client: Client,
) -> Optional[HealthState]:
"""
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
HealthState
"""
return sync_detailed(
client=client,
).parsed
async def asyncio_detailed(
*,
client: Client,
) -> Response[HealthState]:
"""
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[HealthState]
"""
kwargs = _get_kwargs(
client=client,
)
async with httpx.AsyncClient(verify=client.verify_ssl) as _client:
response = await _client.request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
*,
client: Client,
) -> Optional[HealthState]:
"""
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
HealthState
"""
return (
await asyncio_detailed(
client=client,
)
).parsed

View File

@ -0,0 +1,66 @@
import ssl
from typing import Dict, Union
import attr
@attr.s(auto_attribs=True)
class Client:
"""A class for keeping track of data related to the API
Attributes:
base_url: The base URL for the API, all requests are made to a relative path to this URL
cookies: A dictionary of cookies to be sent with every request
headers: A dictionary of headers to be sent with every request
timeout: The maximum amount of a time in seconds a request can take. API functions will raise
httpx.TimeoutException if this is exceeded.
verify_ssl: Whether or not to verify the SSL certificate of the API server. This should be True in production,
but can be set to False for testing purposes.
raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
status code that was not documented in the source OpenAPI document.
follow_redirects: Whether or not to follow redirects. Default value is False.
"""
base_url: str
cookies: Dict[str, str] = attr.ib(factory=dict, kw_only=True)
headers: Dict[str, str] = attr.ib(factory=dict, kw_only=True)
timeout: float = attr.ib(5.0, kw_only=True)
verify_ssl: Union[str, bool, ssl.SSLContext] = attr.ib(True, kw_only=True)
raise_on_unexpected_status: bool = attr.ib(False, kw_only=True)
follow_redirects: bool = attr.ib(False, kw_only=True)
def get_headers(self) -> Dict[str, str]:
"""Get headers to be used in all endpoints"""
return {**self.headers}
def with_headers(self, headers: Dict[str, str]) -> "Client":
"""Get a new client matching this one with additional headers"""
return attr.evolve(self, headers={**self.headers, **headers})
def get_cookies(self) -> Dict[str, str]:
return {**self.cookies}
def with_cookies(self, cookies: Dict[str, str]) -> "Client":
"""Get a new client matching this one with additional cookies"""
return attr.evolve(self, cookies={**self.cookies, **cookies})
def get_timeout(self) -> float:
return self.timeout
def with_timeout(self, timeout: float) -> "Client":
"""Get a new client matching this one with a new timeout (in seconds)"""
return attr.evolve(self, timeout=timeout)
@attr.s(auto_attribs=True)
class AuthenticatedClient(Client):
"""A Client which has been authenticated for use on secured endpoints"""
token: str
prefix: str = "Bearer"
auth_header_name: str = "Authorization"
def get_headers(self) -> Dict[str, str]:
"""Get headers to be used in authenticated endpoints"""
auth_header_value = f"{self.prefix} {self.token}" if self.prefix else self.token
return {self.auth_header_name: auth_header_value, **self.headers}

View File

@ -0,0 +1,14 @@
""" Contains shared errors types that can be raised from API functions """
class UnexpectedStatus(Exception):
"""Raised by api functions when the response status an undocumented status and Client.raise_on_unexpected_status is True"""
def __init__(self, status_code: int, content: bytes):
self.status_code = status_code
self.content = content
super().__init__(f"Unexpected status code: {status_code}")
__all__ = ["UnexpectedStatus"]

View File

@ -0,0 +1,17 @@
""" Contains all the data models used in inputs/outputs """
from .choice import Choice
from .completion_request import CompletionRequest
from .completion_response import CompletionResponse
from .health_state import HealthState
from .log_event_request import LogEventRequest
from .segments import Segments
__all__ = (
"Choice",
"CompletionRequest",
"CompletionResponse",
"HealthState",
"LogEventRequest",
"Segments",
)

View File

@ -0,0 +1,64 @@
from typing import Any, Dict, List, Type, TypeVar
import attr
T = TypeVar("T", bound="Choice")
@attr.s(auto_attribs=True)
class Choice:
"""
Attributes:
index (int):
text (str):
"""
index: int
text: str
additional_properties: Dict[str, Any] = attr.ib(init=False, factory=dict)
def to_dict(self) -> Dict[str, Any]:
index = self.index
text = self.text
field_dict: Dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"index": index,
"text": text,
}
)
return field_dict
@classmethod
def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
d = src_dict.copy()
index = d.pop("index")
text = d.pop("text")
choice = cls(
index=index,
text=text,
)
choice.additional_properties = d
return choice
@property
def additional_keys(self) -> List[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties

View File

@ -0,0 +1,102 @@
from typing import TYPE_CHECKING, Any, Dict, List, Type, TypeVar, Union
import attr
from ..types import UNSET, Unset
if TYPE_CHECKING:
from ..models.segments import Segments
T = TypeVar("T", bound="CompletionRequest")
@attr.s(auto_attribs=True)
class CompletionRequest:
r"""
Example:
{'language': 'python', 'segments': {'prefix': 'def fib(n):\n ', 'suffix': '\n return fib(n - 1) +
fib(n - 2)'}}
Attributes:
prompt (Union[Unset, None, str]): Example: def fib(n):.
language (Union[Unset, None, str]): Language identifier, full list is maintained at
https://code.visualstudio.com/docs/languages/identifiers Example: python.
segments (Union[Unset, None, Segments]):
user (Union[Unset, None, str]):
"""
prompt: Union[Unset, None, str] = UNSET
language: Union[Unset, None, str] = UNSET
segments: Union[Unset, None, "Segments"] = UNSET
user: Union[Unset, None, str] = UNSET
additional_properties: Dict[str, Any] = attr.ib(init=False, factory=dict)
def to_dict(self) -> Dict[str, Any]:
prompt = self.prompt
language = self.language
segments: Union[Unset, None, Dict[str, Any]] = UNSET
if not isinstance(self.segments, Unset):
segments = self.segments.to_dict() if self.segments else None
user = self.user
field_dict: Dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update({})
if prompt is not UNSET:
field_dict["prompt"] = prompt
if language is not UNSET:
field_dict["language"] = language
if segments is not UNSET:
field_dict["segments"] = segments
if user is not UNSET:
field_dict["user"] = user
return field_dict
@classmethod
def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
from ..models.segments import Segments
d = src_dict.copy()
prompt = d.pop("prompt", UNSET)
language = d.pop("language", UNSET)
_segments = d.pop("segments", UNSET)
segments: Union[Unset, None, Segments]
if _segments is None:
segments = None
elif isinstance(_segments, Unset):
segments = UNSET
else:
segments = Segments.from_dict(_segments)
user = d.pop("user", UNSET)
completion_request = cls(
prompt=prompt,
language=language,
segments=segments,
user=user,
)
completion_request.additional_properties = d
return completion_request
@property
def additional_keys(self) -> List[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties

View File

@ -0,0 +1,79 @@
from typing import TYPE_CHECKING, Any, Dict, List, Type, TypeVar
import attr
if TYPE_CHECKING:
from ..models.choice import Choice
T = TypeVar("T", bound="CompletionResponse")
@attr.s(auto_attribs=True)
class CompletionResponse:
"""
Attributes:
id (str):
choices (List['Choice']):
"""
id: str
choices: List["Choice"]
additional_properties: Dict[str, Any] = attr.ib(init=False, factory=dict)
def to_dict(self) -> Dict[str, Any]:
id = self.id
choices = []
for choices_item_data in self.choices:
choices_item = choices_item_data.to_dict()
choices.append(choices_item)
field_dict: Dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"id": id,
"choices": choices,
}
)
return field_dict
@classmethod
def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
from ..models.choice import Choice
d = src_dict.copy()
id = d.pop("id")
choices = []
_choices = d.pop("choices")
for choices_item_data in _choices:
choices_item = Choice.from_dict(choices_item_data)
choices.append(choices_item)
completion_response = cls(
id=id,
choices=choices,
)
completion_response.additional_properties = d
return completion_response
@property
def additional_keys(self) -> List[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties

View File

@ -0,0 +1,71 @@
from typing import Any, Dict, List, Type, TypeVar
import attr
T = TypeVar("T", bound="HealthState")
@attr.s(auto_attribs=True)
class HealthState:
"""
Attributes:
model (str):
device (str):
compute_type (str):
"""
model: str
device: str
compute_type: str
additional_properties: Dict[str, Any] = attr.ib(init=False, factory=dict)
def to_dict(self) -> Dict[str, Any]:
model = self.model
device = self.device
compute_type = self.compute_type
field_dict: Dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"model": model,
"device": device,
"compute_type": compute_type,
}
)
return field_dict
@classmethod
def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
d = src_dict.copy()
model = d.pop("model")
device = d.pop("device")
compute_type = d.pop("compute_type")
health_state = cls(
model=model,
device=device,
compute_type=compute_type,
)
health_state.additional_properties = d
return health_state
@property
def additional_keys(self) -> List[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties

View File

@ -0,0 +1,71 @@
from typing import Any, Dict, List, Type, TypeVar
import attr
T = TypeVar("T", bound="LogEventRequest")
@attr.s(auto_attribs=True)
class LogEventRequest:
"""
Attributes:
type (str): Event type, should be `view` or `select`. Example: view.
completion_id (str):
choice_index (int):
"""
type: str
completion_id: str
choice_index: int
additional_properties: Dict[str, Any] = attr.ib(init=False, factory=dict)
def to_dict(self) -> Dict[str, Any]:
type = self.type
completion_id = self.completion_id
choice_index = self.choice_index
field_dict: Dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"type": type,
"completion_id": completion_id,
"choice_index": choice_index,
}
)
return field_dict
@classmethod
def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
d = src_dict.copy()
type = d.pop("type")
completion_id = d.pop("completion_id")
choice_index = d.pop("choice_index")
log_event_request = cls(
type=type,
completion_id=completion_id,
choice_index=choice_index,
)
log_event_request.additional_properties = d
return log_event_request
@property
def additional_keys(self) -> List[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties

View File

@ -0,0 +1,67 @@
from typing import Any, Dict, List, Type, TypeVar, Union
import attr
from ..types import UNSET, Unset
T = TypeVar("T", bound="Segments")
@attr.s(auto_attribs=True)
class Segments:
"""
Attributes:
prefix (str): Content that appears before the cursor in the editor window.
suffix (Union[Unset, None, str]): Content that appears after the cursor in the editor window.
"""
prefix: str
suffix: Union[Unset, None, str] = UNSET
additional_properties: Dict[str, Any] = attr.ib(init=False, factory=dict)
def to_dict(self) -> Dict[str, Any]:
prefix = self.prefix
suffix = self.suffix
field_dict: Dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"prefix": prefix,
}
)
if suffix is not UNSET:
field_dict["suffix"] = suffix
return field_dict
@classmethod
def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
d = src_dict.copy()
prefix = d.pop("prefix")
suffix = d.pop("suffix", UNSET)
segments = cls(
prefix=prefix,
suffix=suffix,
)
segments.additional_properties = d
return segments
@property
def additional_keys(self) -> List[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties

View File

@ -0,0 +1 @@
# Marker file for PEP 561

View File

@ -0,0 +1,44 @@
""" Contains some shared types for properties """
from http import HTTPStatus
from typing import BinaryIO, Generic, Literal, MutableMapping, Optional, Tuple, TypeVar
import attr
class Unset:
def __bool__(self) -> Literal[False]:
return False
UNSET: Unset = Unset()
FileJsonType = Tuple[Optional[str], BinaryIO, Optional[str]]
@attr.s(auto_attribs=True)
class File:
"""Contains information for file uploads"""
payload: BinaryIO
file_name: Optional[str] = None
mime_type: Optional[str] = None
def to_tuple(self) -> FileJsonType:
"""Return a tuple representation that httpx will accept for multipart/form-data"""
return self.file_name, self.payload, self.mime_type
T = TypeVar("T")
@attr.s(auto_attribs=True)
class Response(Generic[T]):
"""A response from an endpoint"""
status_code: HTTPStatus
content: bytes
headers: MutableMapping[str, str]
parsed: Optional[T]
__all__ = ["File", "Response", "FileJsonType"]

View File

@ -0,0 +1,9 @@
#!/bin/bash
set -ex
cd clients
openapi-python-client generate \
--path ../website/static/openapi.json \
--config ../experimental/openapi/python.yaml \
--meta setup

View File

@ -0,0 +1,2 @@
project_name_override: tabby-python-client
package_name_override: tabby_client