# Code generated by builder. DO NOT EDIT.
"""
Kelvin API Client.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Optional, Union, overload
from typing_extensions import Literal
from kelvin.api.base.api_service_model import AsyncApiServiceModel
from kelvin.api.base.data_model import KList
from kelvin.api.base.http_client.base_client import AsyncBaseClient, Response
from ..model import requests, response, responses, type
[docs]
class User(AsyncApiServiceModel):
@overload
@classmethod
async def list_users(
cls,
username: Optional[Sequence[str]] = None,
email: Optional[Sequence[str]] = None,
name: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: bool = True,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def list_users(
cls,
username: Optional[Sequence[str]] = None,
email: Optional[Sequence[str]] = None,
name: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: bool = True,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def list_users(
cls,
username: Optional[Sequence[str]] = None,
email: Optional[Sequence[str]] = None,
name: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.UsersListPaginatedResponseCursor, responses.UsersListPaginatedResponseLimits]: ...
@overload
@classmethod
async def list_users(
cls,
username: Optional[Sequence[str]] = None,
email: Optional[Sequence[str]] = None,
name: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> KList[responses.UserItem]: ...
[docs]
@classmethod
async def list_users(
cls,
username: Optional[Sequence[str]] = None,
email: Optional[Sequence[str]] = None,
name: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
Union[
KList[responses.UserItem],
responses.UsersListPaginatedResponseCursor,
responses.UsersListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a list of Users and its parameters. The list can be optionally filtered and sorted on the server before being returned.
**Permission Required:** `kelvin.permission.users.read`.
``listUsers``: ``GET`` ``/api/v4/users/list``
Args:
username : :obj:`Sequence[str]`
A filter on the list based on the key `username`. The filter is on the
full name only. All values in array will be filtered as `OR`. The
string can only contain lowercase alphanumeric characters and `.`, `_`
or `-` characters.
email : :obj:`Sequence[str]`
A filter on the list based on the key `email`. The search is case
insensitive and will find partial matches as well. All values in array
will be filtered as `OR`.
name : :obj:`Sequence[str]`
A filter on the list based on the User's `first_name` and `last_name`
separated by a space. The filter is on exact matches only and is case
sensitive. All values in array will be filtered as `OR`.
search : :obj:`Sequence[str]`
Search Users by key `first_name`, `last_name` or `email`. All values
in array will be filtered as `OR`. The search is case insensitive and
will find partial matches as well.
pagination_type : :obj:`Literal['limits', 'cursor', 'stream']`
Method of pagination to use for return results where `total_items` is
greater than `page_size`. `cursor` and `limits` will return one `page`
of results, `stream` will return all results. ('limits', 'cursor',
'stream')
page_size : :obj:`int`
Number of objects to be returned in each page. Page size can range
between 1 and 10000 objects.
page : :obj:`int`
An integer for the wanted page of results. Used only with
`pagination_type` set as `limits`.
next : :obj:`str`
An alphanumeric string bookmark to indicate where to start for the
next page. Used only with `pagination_type` set as `cursor`.
previous : :obj:`str`
An alphanumeric string bookmark to indicate where to end for the
previous page. Used only with `pagination_type` set as `cursor`.
direction : :obj:`Literal['asc', 'desc']`
Sorting order according to the `sort_by` parameter. ('asc', 'desc')
nulls : :obj:`Literal['first', 'last']`
Null ordering according to the `sort_by` parameter. Defaults to
`first` for ascending order and `last` for descending order. ('first',
'last')
sort_by : :obj:`Sequence[str]`
"""
result_types = {"200": responses.UsersListPaginatedResponseCursor, "400": response.Error, "401": response.Error}
# override pagination_type
# stream type is only supported for raw responses
if not _get_response and pagination_type == "stream":
pagination_type = "cursor"
_request = cls._prepare_request(
method="GET",
path="/api/v4/users/list",
values={},
params={
"username": username,
"email": email,
"name": name,
"search": search,
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
"nulls": nulls,
"sort_by": sort_by,
},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = await 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,
)
if pagination_type == "limits":
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.UsersListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.UsersListPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="GET",
params=_request.params,
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[responses.UserItem],
)
return result
@overload
@classmethod
async def get_user_me(
cls,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_user_me(
cls,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_user_me(
cls,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.UserMeGet: ...
[docs]
@classmethod
async def get_user_me(
cls,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.UserMeGet, dict[str, object], Response]:
"""Get Current User
**Permission Required:** `n/a`.
``getUserMe``: ``GET`` ``/api/v4/users/me``
"""
result_types = {"200": responses.UserMeGet, "400": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="GET",
path="/api/v4/users/me",
values={},
params={},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = await 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.UserMeGet,
)
@overload
@classmethod
async def list_user_settings(
cls,
names: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: bool = True,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def list_user_settings(
cls,
names: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: bool = True,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def list_user_settings(
cls,
names: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
responses.UserSettingsListPaginatedResponseCursor, responses.UserSettingsListPaginatedResponseLimits
]: ...
@overload
@classmethod
async def list_user_settings(
cls,
names: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> KList[type.UserSetting]: ...
[docs]
@classmethod
async def list_user_settings(
cls,
names: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
Union[
KList[type.UserSetting],
responses.UserSettingsListPaginatedResponseCursor,
responses.UserSettingsListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a list of User Settings and its parameters. The list can be optionally filtered and sorted on the server before being returned.
**Permission Required:** `kelvin.permission.users.read`.
``listUserSettings``: ``GET`` ``/api/v4/users/settings/list``
Args:
names : :obj:`Sequence[str]`
Filter User Setting list based on the key `setting_name`.
search : :obj:`Sequence[str]`
Search User Setting by key `setting_name`. All values in array will be
filtered as `OR`. The search is case insensitive and will find partial
matches as well.
pagination_type : :obj:`Literal['limits', 'cursor', 'stream']`
Method of pagination to use for return results where `total_items` is
greater than `page_size`. `cursor` and `limits` will return one `page`
of results, `stream` will return all results. ('limits', 'cursor',
'stream')
page_size : :obj:`int`
Number of objects to be returned in each page. Page size can range
between 1 and 10000 objects.
page : :obj:`int`
An integer for the wanted page of results. Used only with
`pagination_type` set as `limits`.
next : :obj:`str`
An alphanumeric string bookmark to indicate where to start for the
next page. Used only with `pagination_type` set as `cursor`.
previous : :obj:`str`
An alphanumeric string bookmark to indicate where to end for the
previous page. Used only with `pagination_type` set as `cursor`.
direction : :obj:`Literal['asc', 'desc']`
Sorting order according to the `sort_by` parameter. ('asc', 'desc')
nulls : :obj:`Literal['first', 'last']`
Null ordering according to the `sort_by` parameter. Defaults to
`first` for ascending order and `last` for descending order. ('first',
'last')
sort_by : :obj:`Sequence[str]`
"""
result_types = {
"200": responses.UserSettingsListPaginatedResponseCursor,
"400": response.Error,
"401": response.Error,
}
# override pagination_type
# stream type is only supported for raw responses
if not _get_response and pagination_type == "stream":
pagination_type = "cursor"
_request = cls._prepare_request(
method="GET",
path="/api/v4/users/settings/list",
values={},
params={
"names": names,
"search": search,
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
"nulls": nulls,
"sort_by": sort_by,
},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = await 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,
)
if pagination_type == "limits":
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.UserSettingsListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.UserSettingsListPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="GET",
params=_request.params,
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[type.UserSetting],
)
return result
@overload
@classmethod
async def delete_user_settings(
cls,
setting_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_user_settings(
cls,
setting_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_user_settings(
cls,
setting_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_user_settings(
cls,
setting_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Permanently delete an existing User Setting. This cannot be undone once the API request has been submitted.
**Permission Required:** `kelvin.permission.users.delete`.
``deleteUserSettings``: ``POST`` ``/api/v4/users/settings/{setting_name}/delete``
Args:
setting_name : :obj:`str`, optional
The User Setting key `setting_name` to delete. The string can only
contain lowercase alphanumeric characters and `.`, `_` or `-`
characters.
"""
result_types = {"200": None, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/users/settings/{setting_name}/delete",
values={"setting_name": setting_name},
params={},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = await 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
async def get_user_settings(
cls,
setting_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_user_settings(
cls,
setting_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_user_settings(
cls,
setting_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.UserSettingsGet: ...
[docs]
@classmethod
async def get_user_settings(
cls,
setting_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.UserSettingsGet, dict[str, object], Response]:
"""Retrieve the parameters of a User Setting.
**Permission Required:** `kelvin.permission.users.read`.
``getUserSettings``: ``GET`` ``/api/v4/users/settings/{setting_name}/get``
Args:
setting_name : :obj:`str`, optional
The User Setting key `setting_name` to get. The string can only
contain lowercase alphanumeric characters and `.`, `_` or `-`
characters.
"""
result_types = {
"200": responses.UserSettingsGet,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="GET",
path="/api/v4/users/settings/{setting_name}/get",
values={"setting_name": setting_name},
params={},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = await 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.UserSettingsGet,
)
@overload
@classmethod
async def update_user_settings(
cls,
setting_name: str,
data: Optional[Union[requests.UserSettingsUpdate, Mapping[str, object]]] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
async def update_user_settings(
cls,
setting_name: str,
data: Optional[Union[requests.UserSettingsUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def update_user_settings(
cls,
setting_name: str,
data: Optional[Union[requests.UserSettingsUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.UserSettingsUpdate: ...
[docs]
@classmethod
async def update_user_settings(
cls,
setting_name: str,
data: Optional[Union[requests.UserSettingsUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.UserSettingsUpdate, dict[str, object], Response]:
"""Updates an existing User Setting `payload` with any new values.
**Permission Required:** `kelvin.permission.users.update`.
``updateUserSettings``: ``POST`` ``/api/v4/users/settings/{setting_name}/update``
Args:
setting_name : :obj:`str`, optional
The User Setting key `setting_name`. The string can only contain
lowercase alphanumeric characters and `.`, `_` or `-` characters.
data: requests.UserSettingsUpdate, optional
**kwargs:
Extra parameters for requests.UserSettingsUpdate
- update_user_settings: dict
"""
result_types = {"200": responses.UserSettingsUpdate, "400": response.Error, "401": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/users/settings/{setting_name}/update",
values={"setting_name": setting_name},
params={},
files={},
headers={},
data=data,
body_type=requests.UserSettingsUpdate,
array_body=False,
**kwargs,
)
if _dry_run:
return _request.to_dict()
_response = await 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.UserSettingsUpdate,
)
@overload
@classmethod
async def list_shared_settings(
cls,
names: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: bool = True,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def list_shared_settings(
cls,
names: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: bool = True,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def list_shared_settings(
cls,
names: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
responses.SharedSettingsListPaginatedResponseCursor, responses.SharedSettingsListPaginatedResponseLimits
]: ...
@overload
@classmethod
async def list_shared_settings(
cls,
names: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> KList[type.SharedSetting]: ...
[docs]
@classmethod
async def list_shared_settings(
cls,
names: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
Union[
KList[type.SharedSetting],
responses.SharedSettingsListPaginatedResponseCursor,
responses.SharedSettingsListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a list of Shared Settings and its parameters. The list can be optionally filtered and sorted on the server before being returned.
**Permission Required:** `kelvin.permission.users.read`.
``listSharedSettings``: ``GET`` ``/api/v4/users/shared/settings/list``
Args:
names : :obj:`Sequence[str]`
Filter Shared Setting list based on the key `setting_name`.
search : :obj:`Sequence[str]`
Search Shared Setting by key `setting_name`. All values in array will
be filtered as `OR`. The search is case insensitive and will find
partial matches as well.
pagination_type : :obj:`Literal['limits', 'cursor', 'stream']`
Method of pagination to use for return results where `total_items` is
greater than `page_size`. `cursor` and `limits` will return one `page`
of results, `stream` will return all results. ('limits', 'cursor',
'stream')
page_size : :obj:`int`
Number of objects to be returned in each page. Page size can range
between 1 and 10000 objects.
page : :obj:`int`
An integer for the wanted page of results. Used only with
`pagination_type` set as `limits`.
next : :obj:`str`
An alphanumeric string bookmark to indicate where to start for the
next page. Used only with `pagination_type` set as `cursor`.
previous : :obj:`str`
An alphanumeric string bookmark to indicate where to end for the
previous page. Used only with `pagination_type` set as `cursor`.
direction : :obj:`Literal['asc', 'desc']`
Sorting order according to the `sort_by` parameter. ('asc', 'desc')
nulls : :obj:`Literal['first', 'last']`
Null ordering according to the `sort_by` parameter. Defaults to
`first` for ascending order and `last` for descending order. ('first',
'last')
sort_by : :obj:`Sequence[str]`
"""
result_types = {
"200": responses.SharedSettingsListPaginatedResponseCursor,
"400": response.Error,
"401": response.Error,
}
# override pagination_type
# stream type is only supported for raw responses
if not _get_response and pagination_type == "stream":
pagination_type = "cursor"
_request = cls._prepare_request(
method="GET",
path="/api/v4/users/shared/settings/list",
values={},
params={
"names": names,
"search": search,
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
"nulls": nulls,
"sort_by": sort_by,
},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = await 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,
)
if pagination_type == "limits":
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.SharedSettingsListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.SharedSettingsListPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="GET",
params=_request.params,
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[type.SharedSetting],
)
return result
@overload
@classmethod
async def delete_shared_settings(
cls,
setting_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_shared_settings(
cls,
setting_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_shared_settings(
cls,
setting_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_shared_settings(
cls,
setting_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Permanently delete an existing Shared Setting. This cannot be undone once the API request has been submitted.
**Permission Required:** `kelvin.permission.users.delete`.
``deleteSharedSettings``: ``POST`` ``/api/v4/users/shared/settings/{setting_name}/delete``
Args:
setting_name : :obj:`str`, optional
The Shared Setting key `setting_name` to delete. The string can only
contain lowercase alphanumeric characters and `.`, `_` or `-`
characters.
"""
result_types = {"200": None, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/users/shared/settings/{setting_name}/delete",
values={"setting_name": setting_name},
params={},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = await 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
async def get_shared_settings(
cls,
setting_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_shared_settings(
cls,
setting_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_shared_settings(
cls,
setting_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.SharedSettingsGet: ...
[docs]
@classmethod
async def get_shared_settings(
cls,
setting_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.SharedSettingsGet, dict[str, object], Response]:
"""Retrieve the parameters of a Shared Setting.
**Permission Required:** `kelvin.permission.users.read`.
``getSharedSettings``: ``GET`` ``/api/v4/users/shared/settings/{setting_name}/get``
Args:
setting_name : :obj:`str`, optional
The Shared Setting key `setting_name` to get. The string can only
contain lowercase alphanumeric characters and `.`, `_` or `-`
characters.
"""
result_types = {
"200": responses.SharedSettingsGet,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="GET",
path="/api/v4/users/shared/settings/{setting_name}/get",
values={"setting_name": setting_name},
params={},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = await 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.SharedSettingsGet,
)
@overload
@classmethod
async def update_shared_settings(
cls,
setting_name: str,
data: Optional[Union[requests.SharedSettingsUpdate, Mapping[str, object]]] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
async def update_shared_settings(
cls,
setting_name: str,
data: Optional[Union[requests.SharedSettingsUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def update_shared_settings(
cls,
setting_name: str,
data: Optional[Union[requests.SharedSettingsUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.SharedSettingsUpdate: ...
[docs]
@classmethod
async def update_shared_settings(
cls,
setting_name: str,
data: Optional[Union[requests.SharedSettingsUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.SharedSettingsUpdate, dict[str, object], Response]:
"""Updates an existing Shared Setting `payload` with any new values.
**Permission Required:** `kelvin.permission.users.update`.
``updateSharedSettings``: ``POST`` ``/api/v4/users/shared/settings/{setting_name}/update``
Args:
setting_name : :obj:`str`, optional
The Shared Setting key `setting_name`. The string can only contain
lowercase alphanumeric characters and `.`, `_` or `-` characters.
data: requests.SharedSettingsUpdate, optional
**kwargs:
Extra parameters for requests.SharedSettingsUpdate
- update_shared_settings: dict
"""
result_types = {"200": responses.SharedSettingsUpdate, "400": response.Error, "401": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/users/shared/settings/{setting_name}/update",
values={"setting_name": setting_name},
params={},
files={},
headers={},
data=data,
body_type=requests.SharedSettingsUpdate,
array_body=False,
**kwargs,
)
if _dry_run:
return _request.to_dict()
_response = await 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.SharedSettingsUpdate,
)
@overload
@classmethod
async def get_user(
cls,
user_id: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_user(
cls,
user_id: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_user(
cls,
user_id: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.UserGet: ...
[docs]
@classmethod
async def get_user(
cls,
user_id: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.UserGet, dict[str, object], Response]:
"""Retrieve the parameters of a User.
**Permission Required:** `kelvin.permission.users.read`.
``getUser``: ``GET`` ``/api/v4/users/{user_id}/get``
Args:
user_id : :obj:`str`, optional
The generated UUID for the User to get.
"""
result_types = {"200": responses.UserGet, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="GET",
path="/api/v4/users/{user_id}/get",
values={"user_id": user_id},
params={},
files={},
headers={},
data=None,
body_type=None,
array_body=False,
)
if _dry_run:
return _request.to_dict()
_response = await 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.UserGet,
)