# 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 ControlChange(AsyncApiServiceModel):
@overload
@classmethod
async def get_control_change_clustering(
cls,
data: Optional[Union[requests.ControlChangeClusteringGet, 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_control_change_clustering(
cls,
data: Optional[Union[requests.ControlChangeClusteringGet, 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_control_change_clustering(
cls,
data: Optional[Union[requests.ControlChangeClusteringGet, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> KList[responses.ControlChangeClustering]: ...
[docs]
@classmethod
async def get_control_change_clustering(
cls,
data: Optional[Union[requests.ControlChangeClusteringGet, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[KList[responses.ControlChangeClustering], dict[str, object], Response]:
"""Retrieve the total count of Control Changes matching an array of `resources` and filter options between two dates grouped by the parameter `time_bucket`. Will also return a list of all the Control Change `id`s counted.
**Permission Required:** `kelvin.permission.control_change.read`.
``getControlChangeClustering``: ``POST`` ``/api/v4/control-changes/clustering/get``
Args:
data: requests.ControlChangeClusteringGet, optional
**kwargs:
Extra parameters for requests.ControlChangeClusteringGet
- get_control_change_clustering: dict
"""
result_types = {
"200": list[responses.ControlChangeClustering],
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/control-changes/clustering/get",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.ControlChangeClusteringGet,
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.ControlChangeClustering],
)
@overload
@classmethod
async def create_control_change(
cls,
data: Optional[Union[requests.ControlChangeCreate, 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_control_change(
cls,
data: Optional[Union[requests.ControlChangeCreate, 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_control_change(
cls,
data: Optional[Union[requests.ControlChangeCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.ControlChangeCreate: ...
[docs]
@classmethod
async def create_control_change(
cls,
data: Optional[Union[requests.ControlChangeCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.ControlChangeCreate, dict[str, object], Response]:
"""Send a new value to be written to an Asset / Data Stream pair. The Control Change Manager will automatically find which Cluster and Bridge to communicate the change to.
**Permission Required:** `kelvin.permission.control_change.create`.
``createControlChange``: ``POST`` ``/api/v4/control-changes/create``
Args:
data: requests.ControlChangeCreate, optional
**kwargs:
Extra parameters for requests.ControlChangeCreate
- create_control_change: dict
"""
result_types = {"201": responses.ControlChangeCreate, "400": response.Error, "401": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/control-changes/create",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.ControlChangeCreate,
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.ControlChangeCreate,
)
@overload
@classmethod
async def get_control_change_last(
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,
data: Optional[Union[requests.ControlChangeLastGet, 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_control_change_last(
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,
data: Optional[Union[requests.ControlChangeLastGet, 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_control_change_last(
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,
data: Optional[Union[requests.ControlChangeLastGet, 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.ControlChangeLastGetPaginatedResponseCursor, responses.ControlChangeLastGetPaginatedResponseLimits
]: ...
@overload
@classmethod
async def get_control_change_last(
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,
data: Optional[Union[requests.ControlChangeLastGet, 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[responses.ControlChangeGet]: ...
[docs]
@classmethod
async def get_control_change_last(
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,
data: Optional[Union[requests.ControlChangeLastGet, Mapping[str, object]]] = None,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[
Union[
KList[responses.ControlChangeGet],
responses.ControlChangeLastGetPaginatedResponseCursor,
responses.ControlChangeLastGetPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a dictionary with a data property containing an array of latest Control Change objects. Only the latest Control Changes for each `resource` in the request filters will be returned.
**Permission Required:** `kelvin.permission.control_change.read`.
``getControlChangeLast``: ``POST`` ``/api/v4/control-changes/last/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')
sort_by : :obj:`Sequence[str]`
data: requests.ControlChangeLastGet, optional
**kwargs:
Extra parameters for requests.ControlChangeLastGet
- get_control_change_last: dict
"""
result_types = {
"200": responses.ControlChangeLastGetPaginatedResponseCursor,
"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/control-changes/last/get",
values={},
params={
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
"sort_by": sort_by,
},
files={},
headers={},
data=data,
body_type=requests.ControlChangeLastGet,
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.ControlChangeLastGetPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.ControlChangeLastGetPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="POST",
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[responses.ControlChangeGet],
)
return result
@overload
@classmethod
async def list_control_changes(
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,
data: Optional[Union[requests.ControlChangesList, 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_control_changes(
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,
data: Optional[Union[requests.ControlChangesList, 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_control_changes(
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,
data: Optional[Union[requests.ControlChangesList, 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.ControlChangesListPaginatedResponseCursor, responses.ControlChangesListPaginatedResponseLimits
]: ...
@overload
@classmethod
async def list_control_changes(
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,
data: Optional[Union[requests.ControlChangesList, 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[responses.ControlChangeGet]: ...
[docs]
@classmethod
async def list_control_changes(
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,
data: Optional[Union[requests.ControlChangesList, Mapping[str, object]]] = None,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[
Union[
KList[responses.ControlChangeGet],
responses.ControlChangesListPaginatedResponseCursor,
responses.ControlChangesListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a list of Control Change objects for specific Asset / Data Stream pairs or Control Change IDs. The list can be optionally filtered and sorted on the server before being returned.
**Permission Required:** `kelvin.permission.control_change.read`.
``listControlChanges``: ``POST`` ``/api/v4/control-changes/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]`
data: requests.ControlChangesList, optional
**kwargs:
Extra parameters for requests.ControlChangesList
- list_control_changes: dict
"""
result_types = {
"200": responses.ControlChangesListPaginatedResponseCursor,
"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/control-changes/list",
values={},
params={
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
"sort_by": sort_by,
},
files={},
headers={},
data=data,
body_type=requests.ControlChangesList,
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.ControlChangesListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.ControlChangesListPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="POST",
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[responses.ControlChangeGet],
)
return result
@overload
@classmethod
async def get_control_change_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,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.ControlChangeRangeGet, 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_control_change_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,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.ControlChangeRangeGet, 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_control_change_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,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.ControlChangeRangeGet, 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.ControlChangeRangeGetPaginatedResponseCursor, responses.ControlChangeRangeGetPaginatedResponseLimits
]: ...
@overload
@classmethod
async def get_control_change_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,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.ControlChangeRangeGet, 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[responses.ControlChangeGet]: ...
[docs]
@classmethod
async def get_control_change_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,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.ControlChangeRangeGet, Mapping[str, object]]] = None,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[
Union[
KList[responses.ControlChangeGet],
responses.ControlChangeRangeGetPaginatedResponseCursor,
responses.ControlChangeRangeGetPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a dictionary with a data property containing an array of Control Changes within a specified time range for all of the `resources` in the `resources` array that match the filter options.
**Permission Required:** `kelvin.permission.control_change.read`.
``getControlChangeRange``: ``POST`` ``/api/v4/control-changes/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')
sort_by : :obj:`Sequence[str]`
data: requests.ControlChangeRangeGet, optional
**kwargs:
Extra parameters for requests.ControlChangeRangeGet
- get_control_change_range: dict
"""
result_types = {
"200": responses.ControlChangeRangeGetPaginatedResponseCursor,
"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/control-changes/range/get",
values={},
params={
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
"sort_by": sort_by,
},
files={},
headers={},
data=data,
body_type=requests.ControlChangeRangeGet,
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.ControlChangeRangeGetPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.ControlChangeRangeGetPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="POST",
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[responses.ControlChangeGet],
)
return result
@overload
@classmethod
async def delete_control_change(
cls,
control_change_id: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_control_change(
cls,
control_change_id: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_control_change(
cls,
control_change_id: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_control_change(
cls,
control_change_id: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Deletes a control change
**Permission Required:** `kelvin.permission.control_change.delete`.
``deleteControlChange``: ``POST`` ``/api/v4/control-changes/{control_change_id}/delete``
Args:
control_change_id : :obj:`str`, optional
A unique randomly generated UUID as the key `id` for the Control
Change.
"""
result_types = {"200": None, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/control-changes/{control_change_id}/delete",
values={"control_change_id": control_change_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_control_change(
cls,
control_change_id: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_control_change(
cls,
control_change_id: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_control_change(
cls,
control_change_id: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.ControlChangeGet: ...
[docs]
@classmethod
async def get_control_change(
cls,
control_change_id: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.ControlChangeGet, dict[str, object], Response]:
"""Retrieves the properties, status and all associated logs of a Control Change.
**Permission Required:** `kelvin.permission.control_change.read`.
``getControlChange``: ``GET`` ``/api/v4/control-changes/{control_change_id}/get``
Args:
control_change_id : :obj:`str`, optional
A unique randomly generated UUID as the key `id` for the Control
Change.
"""
result_types = {
"200": responses.ControlChangeGet,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="GET",
path="/api/v4/control-changes/{control_change_id}/get",
values={"control_change_id": control_change_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.ControlChangeGet,
)