# Code generated by builder. DO NOT EDIT.
"""
Kelvin API Client.
"""
from __future__ import annotations
from collections.abc import Mapping
from typing import Optional, Union, overload
from typing_extensions import Literal
from kelvin.api.base.api_service_model import ApiServiceModel
from kelvin.api.base.data_model import KList
from kelvin.api.base.error import ResponseError
from kelvin.api.base.http_client.base_client import Response, SyncBaseClient
from ..model import requests, response, responses
[docs]
class Thread(ApiServiceModel):
@overload
@classmethod
def create_thread(
cls,
data: Optional[Union[requests.ThreadCreate, Mapping[str, object]]] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
def create_thread(
cls,
data: Optional[Union[requests.ThreadCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
def create_thread(
cls,
data: Optional[Union[requests.ThreadCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> responses.ThreadCreate: ...
[docs]
@classmethod
def create_thread(
cls,
data: Optional[Union[requests.ThreadCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.ThreadCreate, dict[str, object], Response]:
"""Create Thread
**Permission Required:** `kelvin.permission.threads.create`.
``createThread``: ``POST`` ``/api/v4/threads/create``
Args:
data: requests.ThreadCreate, optional
**kwargs:
Extra parameters for requests.ThreadCreate
- create_thread: dict
"""
result_types = {
"201": responses.ThreadCreate,
"400": response.Error,
"401": response.Error,
"500": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/threads/create",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.ThreadCreate,
array_body=False,
**kwargs,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
return cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.ThreadCreate,
)
@overload
@classmethod
def list_threads(
cls,
type: Optional[str] = None,
related_to: Optional[str] = None,
user_id: Optional[str] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
def list_threads(
cls,
type: Optional[str] = None,
related_to: Optional[str] = None,
user_id: Optional[str] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
def list_threads(
cls,
type: Optional[str] = None,
related_to: Optional[str] = None,
user_id: Optional[str] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
) -> KList[responses.ThreadsList]: ...
[docs]
@classmethod
def list_threads(
cls,
type: Optional[str] = None,
related_to: Optional[str] = None,
user_id: Optional[str] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> Union[KList[responses.ThreadsList], dict[str, object], Response]:
"""List Threads
**Pagination Sortable Columns:** `thread.id`
**Permission Required:** `kelvin.permission.threads.read`.
``listThreads``: ``GET`` ``/api/v4/threads/list``
Args:
type : :obj:`str`
Filter threads by type
related_to : :obj:`str`
Filter threads by related_to
user_id : :obj:`str`
Filter threads by user_id
"""
result_types = {
"200": list[responses.ThreadsList],
"400": response.Error,
"401": response.Error,
"500": response.Error,
}
_request = cls._prepare_request(
method="GET",
path="/api/v4/threads/list",
values={},
params={"type": type, "related_to": related_to, "user_id": user_id},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
return cls._process_response_list(
response=_response,
result_types=result_types,
result_type=KList[responses.ThreadsList],
)
@overload
@classmethod
def delete_thread(
cls,
thread_id: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
def delete_thread(
cls,
thread_id: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
def delete_thread(
cls,
thread_id: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
def delete_thread(
cls,
thread_id: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Delete Thread
**Permission Required:** `kelvin.permission.threads.delete`.
``deleteThread``: ``POST`` ``/api/v4/threads/{thread_id}/delete``
Args:
thread_id : :obj:`str`, optional
Thread ID
"""
result_types = {"200": None, "400": response.Error, "403": response.Error, "500": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/threads/{thread_id}/delete",
values={"thread_id": thread_id},
params={},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
@overload
@classmethod
def update_thread_follow(
cls,
thread_id: str,
follow: Optional[bool] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
def update_thread_follow(
cls,
thread_id: str,
follow: Optional[bool] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
def update_thread_follow(
cls,
thread_id: str,
follow: Optional[bool] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
) -> responses.ThreadFollowUpdate: ...
[docs]
@classmethod
def update_thread_follow(
cls,
thread_id: str,
follow: Optional[bool] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> Union[responses.ThreadFollowUpdate, dict[str, object], Response]:
"""Update Thread Follow
**Permission Required:** `kelvin.permission.threads.read`.
``updateThreadFollow``: ``POST`` ``/api/v4/threads/{thread_id}/follow/update``
Args:
thread_id : :obj:`str`, optional
Thread ID
follow : :obj:`bool`
Set user follow value to true or false
"""
result_types = {"200": responses.ThreadFollowUpdate, "404": response.Error, "500": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/threads/{thread_id}/follow/update",
values={"thread_id": thread_id},
params={"follow": follow},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
return cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.ThreadFollowUpdate,
)
@overload
@classmethod
def get_thread(
cls,
thread_id: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
def get_thread(
cls,
thread_id: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
def get_thread(
cls,
thread_id: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
) -> responses.ThreadGet: ...
[docs]
@classmethod
def get_thread(
cls,
thread_id: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> Union[responses.ThreadGet, dict[str, object], Response]:
"""Get Thread
**Permission Required:** `kelvin.permission.threads.read`.
``getThread``: ``GET`` ``/api/v4/threads/{thread_id}/get``
Args:
thread_id : :obj:`str`, optional
Thread ID
"""
result_types = {"200": responses.ThreadGet, "404": response.Error, "500": response.Error}
_request = cls._prepare_request(
method="GET",
path="/api/v4/threads/{thread_id}/get",
values={"thread_id": thread_id},
params={},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
return cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.ThreadGet,
)
@overload
@classmethod
def create_thread_reply(
cls,
thread_id: str,
reply_id: Optional[str] = None,
data: Optional[Union[requests.ThreadReplyCreate, Mapping[str, object]]] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
def create_thread_reply(
cls,
thread_id: str,
reply_id: Optional[str] = None,
data: Optional[Union[requests.ThreadReplyCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
def create_thread_reply(
cls,
thread_id: str,
reply_id: Optional[str] = None,
data: Optional[Union[requests.ThreadReplyCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> responses.ThreadReplyCreate: ...
[docs]
@classmethod
def create_thread_reply(
cls,
thread_id: str,
reply_id: Optional[str] = None,
data: Optional[Union[requests.ThreadReplyCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.ThreadReplyCreate, dict[str, object], Response]:
"""Create Thread Reply
**Permission Required:** `kelvin.permission.threads.create`.
``createThreadReply``: ``POST`` ``/api/v4/threads/{thread_id}/replies/create``
Args:
thread_id : :obj:`str`, optional
Thread ID
reply_id : :obj:`str`
Reply ID
data: requests.ThreadReplyCreate, optional
**kwargs:
Extra parameters for requests.ThreadReplyCreate
- create_thread_reply: dict
"""
result_types = {
"200": responses.ThreadReplyCreate,
"400": response.Error,
"401": response.Error,
"500": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/threads/{thread_id}/replies/create",
values={"thread_id": thread_id},
params={"reply_id": reply_id},
files={},
headers={},
data=data,
body_type=requests.ThreadReplyCreate,
array_body=False,
**kwargs,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
return cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.ThreadReplyCreate,
)
@overload
@classmethod
def delete_thread_reply(
cls,
thread_id: str,
reply_id: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
def delete_thread_reply(
cls,
thread_id: str,
reply_id: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
def delete_thread_reply(
cls,
thread_id: str,
reply_id: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
) -> str: ...
[docs]
@classmethod
def delete_thread_reply(
cls,
thread_id: str,
reply_id: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> Union[str, dict[str, object], Response]:
"""Delete Thread Reply
**Permission Required:** `kelvin.permission.threads.delete`.
``deleteThreadReply``: ``POST`` ``/api/v4/threads/{thread_id}/replies/{reply_id}/delete``
Args:
thread_id : :obj:`str`, optional
Thread ID
reply_id : :obj:`str`, optional
Reply ID
"""
result_types = {"200": str, "400": response.Error, "401": response.Error, "500": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/threads/{thread_id}/replies/{reply_id}/delete",
values={"thread_id": thread_id, "reply_id": reply_id},
params={},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
result = cls._process_str_response(response=_response)
if result is None:
raise ResponseError(
f"Unexpected empty response",
_response,
)
return result
@overload
@classmethod
def update_thread_reply(
cls,
thread_id: str,
reply_id: str,
data: Optional[Union[requests.ThreadReplyUpdate, Mapping[str, object]]] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
def update_thread_reply(
cls,
thread_id: str,
reply_id: str,
data: Optional[Union[requests.ThreadReplyUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
def update_thread_reply(
cls,
thread_id: str,
reply_id: str,
data: Optional[Union[requests.ThreadReplyUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> responses.ThreadReplyUpdate: ...
[docs]
@classmethod
def update_thread_reply(
cls,
thread_id: str,
reply_id: str,
data: Optional[Union[requests.ThreadReplyUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.ThreadReplyUpdate, dict[str, object], Response]:
"""Update Thread Reply
**Permission Required:** `kelvin.permission.threads.update`.
``updateThreadReply``: ``POST`` ``/api/v4/threads/{thread_id}/replies/{reply_id}/update``
Args:
thread_id : :obj:`str`, optional
Thread ID
reply_id : :obj:`str`, optional
Reply ID
data: requests.ThreadReplyUpdate, optional
**kwargs:
Extra parameters for requests.ThreadReplyUpdate
- update_thread_reply: dict
"""
result_types = {
"200": responses.ThreadReplyUpdate,
"400": response.Error,
"401": response.Error,
"500": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/threads/{thread_id}/replies/{reply_id}/update",
values={"thread_id": thread_id, "reply_id": reply_id},
params={},
files={},
headers={},
data=data,
body_type=requests.ThreadReplyUpdate,
array_body=False,
**kwargs,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
return cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.ThreadReplyUpdate,
)
@overload
@classmethod
def update_thread_seen(
cls,
thread_id: str,
seen: Optional[bool] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
def update_thread_seen(
cls,
thread_id: str,
seen: Optional[bool] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
def update_thread_seen(
cls,
thread_id: str,
seen: Optional[bool] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
) -> responses.ThreadSeenUpdate: ...
[docs]
@classmethod
def update_thread_seen(
cls,
thread_id: str,
seen: Optional[bool] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
) -> Union[responses.ThreadSeenUpdate, dict[str, object], Response]:
"""Update Thread Seen
**Permission Required:** `kelvin.permission.threads.read`.
``updateThreadSeen``: ``POST`` ``/api/v4/threads/{thread_id}/seen/update``
Args:
thread_id : :obj:`str`, optional
Thread ID
seen : :obj:`bool`
Set user seen value to true or false
"""
result_types = {"200": responses.ThreadSeenUpdate, "404": response.Error, "500": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/threads/{thread_id}/seen/update",
values={"thread_id": thread_id},
params={"seen": seen},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
return cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.ThreadSeenUpdate,
)
@overload
@classmethod
def update_thread(
cls,
thread_id: str,
data: Optional[Union[requests.ThreadUpdate, Mapping[str, object]]] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
def update_thread(
cls,
thread_id: str,
data: Optional[Union[requests.ThreadUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
def update_thread(
cls,
thread_id: str,
data: Optional[Union[requests.ThreadUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> responses.ThreadUpdate: ...
[docs]
@classmethod
def update_thread(
cls,
thread_id: str,
data: Optional[Union[requests.ThreadUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.ThreadUpdate, dict[str, object], Response]:
"""Update Thread
**Permission Required:** `kelvin.permission.threads.update`.
``updateThread``: ``POST`` ``/api/v4/threads/{thread_id}/update``
Args:
thread_id : :obj:`str`, optional
Thread ID
data: requests.ThreadUpdate, optional
**kwargs:
Extra parameters for requests.ThreadUpdate
- update_thread: dict
"""
result_types = {
"200": responses.ThreadUpdate,
"400": response.Error,
"401": response.Error,
"500": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/threads/{thread_id}/update",
values={"thread_id": thread_id},
params={},
files={},
headers={},
data=data,
body_type=requests.ThreadUpdate,
array_body=False,
**kwargs,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
return cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.ThreadUpdate,
)