# 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
[docs]
class Secret(AsyncApiServiceModel):
@overload
@classmethod
async def create_secret(
cls,
data: Optional[Union[requests.SecretCreate, 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 create_secret(
cls,
data: Optional[Union[requests.SecretCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def create_secret(
cls,
data: Optional[Union[requests.SecretCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.SecretCreate: ...
[docs]
@classmethod
async def create_secret(
cls,
data: Optional[Union[requests.SecretCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.SecretCreate, dict[str, object], Response]:
"""Create a new Secret.
Once this is created you can not change or see the value itself from Kelvin API. Retrieval of the value can only be done through an App.
**Permission Required:** `kelvin.permission.secret.create`.
``createSecret``: ``POST`` ``/api/v4/secrets/create``
Args:
data: requests.SecretCreate, optional
**kwargs:
Extra parameters for requests.SecretCreate
- create_secret: dict
"""
result_types = {
"201": responses.SecretCreate,
"400": response.Error,
"401": response.Error,
"409": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/secrets/create",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.SecretCreate,
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.SecretCreate,
)
@overload
@classmethod
async def list_secrets(
cls,
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,
sort_by: Optional[Sequence[str]] = None,
search: 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_secrets(
cls,
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,
sort_by: Optional[Sequence[str]] = None,
search: 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_secrets(
cls,
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,
sort_by: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.SecretsListPaginatedResponseCursor, responses.SecretsListPaginatedResponseLimits]: ...
@overload
@classmethod
async def list_secrets(
cls,
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,
sort_by: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> KList[responses.SecretItem]: ...
[docs]
@classmethod
async def list_secrets(
cls,
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,
sort_by: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
Union[
KList[responses.SecretItem],
responses.SecretsListPaginatedResponseCursor,
responses.SecretsListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a list of Secrets. The actual Secret itself can not be retrieved here and is only available from an App.
**Permission Required:** `kelvin.permission.secret.read`.
``listSecrets``: ``GET`` ``/api/v4/secrets/list``
Args:
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')
sort_by : :obj:`Sequence[str]`
search : :obj:`Sequence[str]`
"""
result_types = {
"200": responses.SecretsListPaginatedResponseCursor,
"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/secrets/list",
values={},
params={
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
"sort_by": sort_by,
"search": search,
},
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.SecretsListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.SecretsListPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="GET",
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[responses.SecretItem],
)
return result
@overload
@classmethod
async def delete_secret(
cls,
secret_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_secret(
cls,
secret_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_secret(
cls,
secret_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_secret(
cls,
secret_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Permanently delete a Secret. This cannot be undone once the API request has been submitted.
**Permission Required:** `kelvin.permission.secret.delete`.
``deleteSecret``: ``POST`` ``/api/v4/secrets/{secret_name}/delete``
Args:
secret_name : :obj:`str`, optional
Secret key `name` to delete. The string can only contain lowercase
alphanumeric characters and `-` characters.
"""
result_types = {"200": None, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/secrets/{secret_name}/delete",
values={"secret_name": secret_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 update_secret(
cls,
secret_name: str,
data: Optional[Union[requests.SecretUpdate, 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_secret(
cls,
secret_name: str,
data: Optional[Union[requests.SecretUpdate, 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_secret(
cls,
secret_name: str,
data: Optional[Union[requests.SecretUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.SecretUpdate: ...
[docs]
@classmethod
async def update_secret(
cls,
secret_name: str,
data: Optional[Union[requests.SecretUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.SecretUpdate, dict[str, object], Response]:
"""Update an existing Secret with a new 'value'. Any parameters that are not provided will remain unchanged.
**Permission Required:** `kelvin.permission.secret.update`.
``updateSecret``: ``POST`` ``/api/v4/secrets/{secret_name}/update``
Args:
secret_name : :obj:`str`, optional
Secret key `name` to update. The string can only contain lowercase
alphanumeric characters and `-` characters.
data: requests.SecretUpdate, optional
**kwargs:
Extra parameters for requests.SecretUpdate
- update_secret: dict
"""
result_types = {
"200": responses.SecretUpdate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/secrets/{secret_name}/update",
values={"secret_name": secret_name},
params={},
files={},
headers={},
data=data,
body_type=requests.SecretUpdate,
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.SecretUpdate,
)