# Code generated by builder. DO NOT EDIT.
"""
Kelvin API Client.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from datetime import datetime
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 AsyncKIterator, KList
from kelvin.api.base.error import ResponseError
from kelvin.api.base.http_client.base_client import AsyncBaseClient, Response
from ..model import requests, response, responses, type
[docs]
class CustomActions(AsyncApiServiceModel):
@overload
@classmethod
async def get_custom_actions_clustering(
cls,
data: Optional[Union[requests.CustomActionsClusteringGet, 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 get_custom_actions_clustering(
cls,
data: Optional[Union[requests.CustomActionsClusteringGet, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def get_custom_actions_clustering(
cls,
data: Optional[Union[requests.CustomActionsClusteringGet, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> KList[responses.CustomActionsClustering]: ...
[docs]
@classmethod
async def get_custom_actions_clustering(
cls,
data: Optional[Union[requests.CustomActionsClusteringGet, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[KList[responses.CustomActionsClustering], dict[str, object], Response]:
"""Retrieve the number of Custom Actions matching the filters between two dates, grouped by the
`time_bucket` period. Each bucket also returns the Custom Action `id`s counted and a per-`state`
breakdown of the count.
**Permission Required:** `kelvin.permission.custom-actions.read`.
``getCustomActionsClustering``: ``POST`` ``/api/v4/custom-actions/clustering/get``
Args:
data: requests.CustomActionsClusteringGet, optional
**kwargs:
Extra parameters for requests.CustomActionsClusteringGet
- get_custom_actions_clustering: dict
"""
result_types = {"200": list[responses.CustomActionsClustering], "400": response.Error, "401": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/custom-actions/clustering/get",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.CustomActionsClusteringGet,
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_list(
response=_response,
result_types=result_types,
result_type=KList[responses.CustomActionsClustering],
)
@overload
@classmethod
async def create_custom_action(
cls,
data: Optional[Union[requests.CustomActionCreate, 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_custom_action(
cls,
data: Optional[Union[requests.CustomActionCreate, 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_custom_action(
cls,
data: Optional[Union[requests.CustomActionCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.CustomActionCreate: ...
[docs]
@classmethod
async def create_custom_action(
cls,
data: Optional[Union[requests.CustomActionCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.CustomActionCreate, dict[str, object], Response]:
"""Create Custom Action.
The platform validates `payload` against the schema bound to `type` before
creating the Custom Action. Types without a schema accept any object payload.
**Permission Required:** `kelvin.permission.custom-actions.create`.
``createCustomAction``: ``POST`` ``/api/v4/custom-actions/create``
Args:
data: requests.CustomActionCreate, optional
**kwargs:
Extra parameters for requests.CustomActionCreate
- create_custom_action: dict
"""
result_types = {
"201": responses.CustomActionCreate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
"409": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/custom-actions/create",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.CustomActionCreate,
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.CustomActionCreate,
)
@overload
@classmethod
async def list_custom_actions(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.CustomActionsList, Mapping[str, object]]] = None,
fetch: bool = True,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
async def list_custom_actions(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.CustomActionsList, Mapping[str, object]]] = None,
fetch: bool = True,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def list_custom_actions(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.CustomActionsList, Mapping[str, object]]] = None,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[
responses.CustomActionsListPaginatedResponseCursor, responses.CustomActionsListPaginatedResponseLimits
]: ...
@overload
@classmethod
async def list_custom_actions(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.CustomActionsList, Mapping[str, object]]] = None,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> KList[type.CustomAction]: ...
[docs]
@classmethod
async def list_custom_actions(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.CustomActionsList, Mapping[str, object]]] = None,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[
Union[
KList[type.CustomAction],
responses.CustomActionsListPaginatedResponseCursor,
responses.CustomActionsListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns Custom Actions that match the provided filters.
Use query parameters for pagination and sorting, and the request body for
field-based filtering.
**Permission Required:** `kelvin.permission.custom-actions.read`.
``listCustomActions``: ``POST`` ``/api/v4/custom-actions/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')
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]`
data: requests.CustomActionsList, optional
**kwargs:
Extra parameters for requests.CustomActionsList
- list_custom_actions: dict
"""
result_types = {
"200": responses.CustomActionsListPaginatedResponseCursor,
"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="POST",
path="/api/v4/custom-actions/list",
values={},
params={
"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=data,
body_type=requests.CustomActionsList,
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,
)
if pagination_type == "limits":
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.CustomActionsListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.CustomActionsListPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="POST",
params=_request.params,
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[type.CustomAction],
)
return result
@overload
@classmethod
async def get_custom_actions_range(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.CustomActionsRangeGet, Mapping[str, object]]] = None,
fetch: bool = True,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_custom_actions_range(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.CustomActionsRangeGet, Mapping[str, object]]] = None,
fetch: bool = True,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def get_custom_actions_range(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.CustomActionsRangeGet, Mapping[str, object]]] = None,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[
responses.CustomActionsRangeGetPaginatedResponseCursor, responses.CustomActionsRangeGetPaginatedResponseLimits
]: ...
@overload
@classmethod
async def get_custom_actions_range(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.CustomActionsRangeGet, Mapping[str, object]]] = None,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> KList[type.CustomAction]: ...
[docs]
@classmethod
async def get_custom_actions_range(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.CustomActionsRangeGet, Mapping[str, object]]] = None,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[
Union[
KList[type.CustomAction],
responses.CustomActionsRangeGetPaginatedResponseCursor,
responses.CustomActionsRangeGetPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns Custom Actions whose `timestamp` falls within the requested range (`start_date` inclusive,
`end_date` exclusive) and that match the request filters. Supports the same pagination and sorting as
the list endpoint.
**Permission Required:** `kelvin.permission.custom-actions.read`.
``getCustomActionsRange``: ``POST`` ``/api/v4/custom-actions/range/get``
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')
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]`
data: requests.CustomActionsRangeGet, optional
**kwargs:
Extra parameters for requests.CustomActionsRangeGet
- get_custom_actions_range: dict
"""
result_types = {
"200": responses.CustomActionsRangeGetPaginatedResponseCursor,
"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="POST",
path="/api/v4/custom-actions/range/get",
values={},
params={
"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=data,
body_type=requests.CustomActionsRangeGet,
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,
)
if pagination_type == "limits":
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.CustomActionsRangeGetPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.CustomActionsRangeGetPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="POST",
params=_request.params,
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[type.CustomAction],
)
return result
@overload
@classmethod
async def list_custom_actions_sync_final(
cls,
cluster: str,
since: datetime,
sort_by: Optional[Sequence[str]] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def list_custom_actions_sync_final(
cls,
cluster: str,
since: datetime,
sort_by: Optional[Sequence[str]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def list_custom_actions_sync_final(
cls,
cluster: str,
since: datetime,
sort_by: Optional[Sequence[str]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> AsyncKIterator[responses.CustomActionsSyncFinalListPaginatedResponseStream]: ...
[docs]
@classmethod
async def list_custom_actions_sync_final(
cls,
cluster: str,
since: datetime,
sort_by: Optional[Sequence[str]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
AsyncKIterator[responses.CustomActionsSyncFinalListPaginatedResponseStream], dict[str, object], Response
]:
"""Returns the list of Custom Actions objects intended for the Apps that
produced them.
The Custom Actions returned here are filtered by:
- Custom Actions which `created_by` is of a workload running on the specified
`cluster`
- Custom Actions which `updated` is after the specified `since`
- Custom Actions whose `state` is a final state: `completed`, `failed`, or `cancelled`
**Permission Required:** `kelvin.permission.custom-actions.read`.
``listCustomActionsSyncFinal``: ``GET`` ``/api/v4/custom-actions/sync/{cluster}/final/list``
Args:
cluster : :obj:`str`, optional
since : :obj:`datetime`, optional
Return only Custom Actions with an `updated` date strictly after this
value.
sort_by : :obj:`Sequence[str]`
"""
result_types = {
"200": responses.CustomActionsSyncFinalListPaginatedResponseStream,
"400": response.Error,
"401": response.Error,
}
_request = cls._prepare_request(
method="GET",
path="/api/v4/custom-actions/sync/{cluster}/final/list",
values={"cluster": cluster},
params={"since": since, "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=True,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
result = await cls._process_response_kiterator(
response=_response,
result_types=result_types,
return_type=responses.CustomActionsSyncFinalListPaginatedResponseStream,
)
if result is None:
raise ResponseError(
f"Unexpected empty response",
_response,
)
return result
@overload
@classmethod
async def get_custom_actions_sync(
cls,
cluster: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_custom_actions_sync(
cls,
cluster: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_custom_actions_sync(
cls,
cluster: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.CustomActionsSyncGet: ...
[docs]
@classmethod
async def get_custom_actions_sync(
cls,
cluster: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.CustomActionsSyncGet, dict[str, object], Response]:
"""Get Custom Actions (Sync).
**Permission Required:** `kelvin.permission.custom-actions.read`.
``getCustomActionsSync``: ``GET`` ``/api/v4/custom-actions/sync/{cluster}/get``
Args:
cluster : :obj:`str`, optional
"""
result_types = {
"200": responses.CustomActionsSyncGet,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="GET",
path="/api/v4/custom-actions/sync/{cluster}/get",
values={"cluster": cluster},
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.CustomActionsSyncGet,
)
@overload
@classmethod
async def list_custom_actions_sync(
cls,
cluster: str,
sort_by: 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,
fetch: bool = True,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def list_custom_actions_sync(
cls,
cluster: str,
sort_by: 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,
fetch: bool = True,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def list_custom_actions_sync(
cls,
cluster: str,
sort_by: 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,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
responses.CustomActionsSyncListPaginatedResponseCursor, responses.CustomActionsSyncListPaginatedResponseLimits
]: ...
@overload
@classmethod
async def list_custom_actions_sync(
cls,
cluster: str,
sort_by: 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,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> KList[type.CustomAction]: ...
[docs]
@classmethod
async def list_custom_actions_sync(
cls,
cluster: str,
sort_by: 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,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
Union[
KList[type.CustomAction],
responses.CustomActionsSyncListPaginatedResponseCursor,
responses.CustomActionsSyncListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a list of Custom Action objects for the specified cluster. The list can be optionally filtered and sorted on the server before being returned.
**Permission Required:** `kelvin.permission.custom-actions.read`.
``listCustomActionsSync``: ``GET`` ``/api/v4/custom-actions/sync/{cluster}/pending/list``
Args:
cluster : :obj:`str`, optional
sort_by : :obj:`Sequence[str]`
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')
"""
result_types = {
"200": responses.CustomActionsSyncListPaginatedResponseCursor,
"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/custom-actions/sync/{cluster}/pending/list",
values={"cluster": cluster},
params={
"sort_by": sort_by,
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
"nulls": nulls,
},
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.CustomActionsSyncListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.CustomActionsSyncListPaginatedResponseCursor,
)
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.CustomAction],
)
return result
@overload
@classmethod
async def create_custom_actions_type(
cls,
data: Optional[Union[requests.CustomActionsTypeCreate, 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_custom_actions_type(
cls,
data: Optional[Union[requests.CustomActionsTypeCreate, 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_custom_actions_type(
cls,
data: Optional[Union[requests.CustomActionsTypeCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.CustomActionsTypeCreate: ...
[docs]
@classmethod
async def create_custom_actions_type(
cls,
data: Optional[Union[requests.CustomActionsTypeCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.CustomActionsTypeCreate, dict[str, object], Response]:
"""Create a new Custom Action Type.
`schema`, when provided, must be a valid JSON Schema for the Custom Action
`payload` object. The platform binds the schema to the new type and uses it
to validate payloads on `/custom-actions/create`. To change the payload
contract, create a new Custom Action type.
**Permission Required:** `kelvin.permission.custom-actions.create`.
``createCustomActionsType``: ``POST`` ``/api/v4/custom-actions/types/create``
Args:
data: requests.CustomActionsTypeCreate, optional
**kwargs:
Extra parameters for requests.CustomActionsTypeCreate
- create_custom_actions_type: dict
"""
result_types = {
"201": responses.CustomActionsTypeCreate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/custom-actions/types/create",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.CustomActionsTypeCreate,
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.CustomActionsTypeCreate,
)
@overload
@classmethod
async def list_custom_actions_types(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
app_names: 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_custom_actions_types(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
app_names: 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_custom_actions_types(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
app_names: Optional[Sequence[str]] = None,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.CustomActionsTypesListPaginatedCursor, responses.CustomActionsTypesListPaginatedLimits]: ...
@overload
@classmethod
async def list_custom_actions_types(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
app_names: Optional[Sequence[str]] = None,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> KList[type.CustomActionType]: ...
[docs]
@classmethod
async def list_custom_actions_types(
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,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
app_names: Optional[Sequence[str]] = None,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
Union[
KList[type.CustomActionType],
responses.CustomActionsTypesListPaginatedCursor,
responses.CustomActionsTypesListPaginatedLimits,
],
dict[str, object],
Response,
]:
"""Returns a list of Custom Action Type objects. The list can be optionally filtered and sorted on the server before being returned.
**Permission Required:** `kelvin.permission.custom-actions.read`.
``listCustomActionsTypes``: ``GET`` ``/api/v4/custom-actions/types/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')
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]`
search : :obj:`Sequence[str]`
Search and filter on the list based on the key `name` and `title`. The
search is case insensitive and will find partial matches as well.
app_names : :obj:`Sequence[str]`
Filter the list of custom action types to include only those that are
currently in use by the specified applications.
"""
result_types = {
"200": responses.CustomActionsTypesListPaginatedCursor,
"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/custom-actions/types/list",
values={},
params={
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
"nulls": nulls,
"sort_by": sort_by,
"search": search,
"app_names": app_names,
},
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.CustomActionsTypesListPaginatedLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.CustomActionsTypesListPaginatedCursor,
)
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.CustomActionType],
)
return result
@overload
@classmethod
async def delete_custom_action_type(
cls,
name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_custom_action_type(
cls,
name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_custom_action_type(
cls,
name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_custom_action_type(
cls,
name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Permanently delete an existing Custom Action Type. An error will be returned if there are any current Custom Action linked to the Custom Action Type. This cannot be undone once the API request has been submitted.
**Permission Required:** `kelvin.permission.custom-actions.delete`.
``deleteCustomActionType``: ``POST`` ``/api/v4/custom-actions/types/{name}/delete``
Args:
name : :obj:`str`, optional
Custom Action Type key `name` to delete. Case sensitive name.
"""
result_types = {
"204": None,
"400": response.Error,
"401": response.Error,
"404": response.Error,
"409": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/custom-actions/types/{name}/delete",
values={"name": 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_custom_actions_type(
cls,
name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_custom_actions_type(
cls,
name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_custom_actions_type(
cls,
name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.CustomActionsTypeGet: ...
[docs]
@classmethod
async def get_custom_actions_type(
cls,
name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.CustomActionsTypeGet, dict[str, object], Response]:
"""Retrieves the parameters of a Custom Action Type.
**Permission Required:** `kelvin.permission.custom-actions.read`.
``getCustomActionsType``: ``GET`` ``/api/v4/custom-actions/types/{name}/get``
Args:
name : :obj:`str`, optional
Custom Action Type key `name` to get. Case sensitive name.
"""
result_types = {
"200": responses.CustomActionsTypeGet,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="GET",
path="/api/v4/custom-actions/types/{name}/get",
values={"name": 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.CustomActionsTypeGet,
)
@overload
@classmethod
async def update_custom_actions_type(
cls,
name: str,
data: Optional[Union[requests.CustomActionsTypeUpdate, 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_custom_actions_type(
cls,
name: str,
data: Optional[Union[requests.CustomActionsTypeUpdate, 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_custom_actions_type(
cls,
name: str,
data: Optional[Union[requests.CustomActionsTypeUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.CustomActionsTypeUpdate: ...
[docs]
@classmethod
async def update_custom_actions_type(
cls,
name: str,
data: Optional[Union[requests.CustomActionsTypeUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.CustomActionsTypeUpdate, dict[str, object], Response]:
"""Updates an existing Custom Action Type with any new values passed through the body parameters. All body parameters are optional and if not provided will remain unchanged. Only the unique identifier `name` can not be changed.
**Permission Required:** `kelvin.permission.custom-actions.update`.
``updateCustomActionsType``: ``POST`` ``/api/v4/custom-actions/types/{name}/update``
Args:
name : :obj:`str`, optional
Custom Action Type key `name` to update. Case sensitive name.
data: requests.CustomActionsTypeUpdate, optional
**kwargs:
Extra parameters for requests.CustomActionsTypeUpdate
- update_custom_actions_type: dict
"""
result_types = {
"200": responses.CustomActionsTypeUpdate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/custom-actions/types/{name}/update",
values={"name": name},
params={},
files={},
headers={},
data=data,
body_type=requests.CustomActionsTypeUpdate,
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.CustomActionsTypeUpdate,
)
@overload
@classmethod
async def delete_custom_action(
cls,
action_id: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_custom_action(
cls,
action_id: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_custom_action(
cls,
action_id: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_custom_action(
cls,
action_id: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Permanently delete an existing Custom Action.
**Permission Required:** `kelvin.permission.custom-actions.delete`.
``deleteCustomAction``: ``POST`` ``/api/v4/custom-actions/{action_id}/delete``
Args:
action_id : :obj:`str`, optional
"""
result_types = {"204": None, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/custom-actions/{action_id}/delete",
values={"action_id": action_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,
)
@overload
@classmethod
async def get_custom_action(
cls,
action_id: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_custom_action(
cls,
action_id: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_custom_action(
cls,
action_id: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.CustomActionGet: ...
[docs]
@classmethod
async def get_custom_action(
cls,
action_id: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.CustomActionGet, dict[str, object], Response]:
"""Returns a single Custom Action by `action_id`.
Includes current lifecycle state, transition history, and payload details.
**Permission Required:** `kelvin.permission.custom-actions.read`.
``getCustomAction``: ``GET`` ``/api/v4/custom-actions/{action_id}/get``
Args:
action_id : :obj:`str`, optional
UUID of the Custom Action to retrieve.
"""
result_types = {
"200": responses.CustomActionGet,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="GET",
path="/api/v4/custom-actions/{action_id}/get",
values={"action_id": action_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.CustomActionGet,
)
@overload
@classmethod
async def update_custom_action(
cls,
action_id: str,
data: Optional[Union[requests.CustomActionUpdate, 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_custom_action(
cls,
action_id: str,
data: Optional[Union[requests.CustomActionUpdate, 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_custom_action(
cls,
action_id: str,
data: Optional[Union[requests.CustomActionUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.CustomActionUpdate: ...
[docs]
@classmethod
async def update_custom_action(
cls,
action_id: str,
data: Optional[Union[requests.CustomActionUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.CustomActionUpdate, dict[str, object], Response]:
"""Updates the Custom Action. The update is absolute, all fields must be set
(or unset) to the intended value.
The Custom Action's state is inferred by the `status_log`. A `response` can
only exist if the state is `completed` or `failed`.
**Permission Required:** `kelvin.permission.custom-actions.update`.
``updateCustomAction``: ``POST`` ``/api/v4/custom-actions/{action_id}/update``
Args:
action_id : :obj:`str`, optional
data: requests.CustomActionUpdate, optional
**kwargs:
Extra parameters for requests.CustomActionUpdate
- update_custom_action: dict
"""
result_types = {
"200": responses.CustomActionUpdate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/custom-actions/{action_id}/update",
values={"action_id": action_id},
params={},
files={},
headers={},
data=data,
body_type=requests.CustomActionUpdate,
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.CustomActionUpdate,
)