# Code generated by builder. DO NOT EDIT.
"""
Kelvin API Client.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Optional, Union, overload
from typing_extensions import Literal
from kelvin.api.base.api_service_model import AsyncApiServiceModel
from kelvin.api.base.data_model import KList
from kelvin.api.base.http_client.base_client import AsyncBaseClient, Response
from ..model import requests, response, responses, type
[docs]
class 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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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 the latest Control Change for each resource that matches the
request filters.
**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')
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.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,
"nulls": nulls,
"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",
params=_request.params,
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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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 Control Changes that match the request filters.
**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')
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.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,
"nulls": nulls,
"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",
params=_request.params,
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[responses.ControlChangeGet],
)
return result
@overload
@classmethod
async def create_control_change_policy(
cls,
data: Optional[Union[requests.ControlChangePolicyCreate, 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_policy(
cls,
data: Optional[Union[requests.ControlChangePolicyCreate, 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_policy(
cls,
data: Optional[Union[requests.ControlChangePolicyCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.ControlChangePolicyCreate: ...
[docs]
@classmethod
async def create_control_change_policy(
cls,
data: Optional[Union[requests.ControlChangePolicyCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.ControlChangePolicyCreate, dict[str, object], Response]:
"""Create a new Control Change Policy.
A Control Change Policy defines default execution behavior and completion
criteria for Control Changes on a resource.
Policy fields override platform defaults. If a field is also set in the
Control Change request, the Control Change value takes precedence.
Resolution is per field: if a field is unset in both the Control Change
request and the policy, it falls back to the platform default.
`expiration_date` is required in Control Change requests and is not part of
policy configuration. Guardrails are configured separately and are not part
of policy configuration.
Validation rules:
- The policy `resource` and all `policy.resource_value_check[].resource`
values must belong to the same asset.
- All referenced datastreams must share the same data type.
**Permission Required:** `kelvin.permission.control_change.create`.
``createControlChangePolicy``: ``POST`` ``/api/v4/control-changes/policies/create``
Args:
data: requests.ControlChangePolicyCreate, optional
**kwargs:
Extra parameters for requests.ControlChangePolicyCreate
- create_control_change_policy: str
"""
result_types = {
"201": responses.ControlChangePolicyCreate,
"400": response.Error,
"401": response.Error,
"409": response.Error,
"500": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/control-changes/policies/create",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.ControlChangePolicyCreate,
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.ControlChangePolicyCreate,
)
@overload
@classmethod
async def list_control_change_policies(
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,
search: Optional[Sequence[str]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.ControlChangePoliciesList, 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_change_policies(
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,
search: Optional[Sequence[str]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.ControlChangePoliciesList, 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_change_policies(
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,
search: Optional[Sequence[str]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.ControlChangePoliciesList, 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.ControlChangePoliciesListPaginatedResponseCursor,
responses.ControlChangePoliciesListPaginatedResponseLimits,
]: ...
@overload
@classmethod
async def list_control_change_policies(
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,
search: Optional[Sequence[str]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.ControlChangePoliciesList, 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.ControlChangePolicy]: ...
[docs]
@classmethod
async def list_control_change_policies(
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,
search: Optional[Sequence[str]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.ControlChangePoliciesList, 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.ControlChangePolicy],
responses.ControlChangePoliciesListPaginatedResponseCursor,
responses.ControlChangePoliciesListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""List Control Change Policies based on the provided filters.
Returned items include only fields explicitly configured in each resource
policy. Fields omitted in policy configuration inherit at runtime based on
per-field precedence: Control Change request value, then resource policy,
then platform defaults.
**Permission Required:** `kelvin.permission.control_change.read`.
``listControlChangePolicies``: ``POST`` ``/api/v4/control-changes/policies/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')
search : :obj:`Sequence[str]`
Search for Control Change Policies with `resources` matching the
search term.
sort_by : :obj:`Sequence[str]`
data: requests.ControlChangePoliciesList, optional
**kwargs:
Extra parameters for requests.ControlChangePoliciesList
- list_control_change_policies: dict
"""
result_types = {
"200": responses.ControlChangePoliciesListPaginatedResponseCursor,
"400": response.Error,
"401": response.Error,
"404": response.Error,
"500": 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/policies/list",
values={},
params={
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
"search": search,
"sort_by": sort_by,
},
files={},
headers={},
data=data,
body_type=requests.ControlChangePoliciesList,
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.ControlChangePoliciesListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.ControlChangePoliciesListPaginatedResponseCursor,
)
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.ControlChangePolicy],
)
return result
@overload
@classmethod
async def delete_control_change_policy(
cls,
resource: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_control_change_policy(
cls,
resource: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_control_change_policy(
cls,
resource: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_control_change_policy(
cls,
resource: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Delete a Control Change Policy by resource.
**Permission Required:** `kelvin.permission.control_change.delete`.
``deleteControlChangePolicy``: ``POST`` ``/api/v4/control-changes/policies/{resource}/delete``
Args:
resource : :obj:`str`, optional
Asset/Data Stream pair used to delete the associated Control Change
Policy.
"""
result_types = {
"200": None,
"400": response.Error,
"401": response.Error,
"404": response.Error,
"500": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/control-changes/policies/{resource}/delete",
values={"resource": resource},
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_policy(
cls,
resource: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_control_change_policy(
cls,
resource: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_control_change_policy(
cls,
resource: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.ControlChangePolicyGet: ...
[docs]
@classmethod
async def get_control_change_policy(
cls,
resource: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.ControlChangePolicyGet, dict[str, object], Response]:
"""Get a Control Change Policy by resource.
The response returns only fields explicitly configured in the resource
policy. Fields omitted in policy configuration inherit at runtime based on
per-field precedence: Control Change request value, then resource policy,
then platform defaults.
`expiration_date` and guardrails are not part of policy configuration.
**Permission Required:** `kelvin.permission.control_change.read`.
``getControlChangePolicy``: ``GET`` ``/api/v4/control-changes/policies/{resource}/get``
Args:
resource : :obj:`str`, optional
Asset/Data Stream pair to retrieve the associated Control Change
Policy.
"""
result_types = {
"200": responses.ControlChangePolicyGet,
"400": response.Error,
"401": response.Error,
"404": response.Error,
"500": response.Error,
}
_request = cls._prepare_request(
method="GET",
path="/api/v4/control-changes/policies/{resource}/get",
values={"resource": resource},
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.ControlChangePolicyGet,
)
@overload
@classmethod
async def update_control_change_policy(
cls,
resource: str,
data: Optional[Union[requests.ControlChangePolicyUpdate, 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_control_change_policy(
cls,
resource: str,
data: Optional[Union[requests.ControlChangePolicyUpdate, 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_control_change_policy(
cls,
resource: str,
data: Optional[Union[requests.ControlChangePolicyUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.ControlChangePolicyUpdate: ...
[docs]
@classmethod
async def update_control_change_policy(
cls,
resource: str,
data: Optional[Union[requests.ControlChangePolicyUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.ControlChangePolicyUpdate, dict[str, object], Response]:
"""Replace the full Control Change Policy configuration for the resource.
Any policy fields omitted from this request are unset in policy
configuration and therefore inherit at runtime, except
`resource_value_check`, which is cleared when omitted.
Policy fields override platform defaults. If a field is also set in the
Control Change request, the Control Change value takes precedence.
Resolution is per field: if a field is unset in both the Control Change
request and the policy, it falls back to the platform default.
`expiration_date` is required in Control Change requests and is not part of
policy configuration. Guardrails are configured separately and are not part
of policy configuration.
Validation rules:
- The path `resource` and all `policy.resource_value_check[].resource`
values must belong to the same asset.
- All referenced datastreams must share the same data type.
**Permission Required:** `kelvin.permission.control_change.update`.
``updateControlChangePolicy``: ``POST`` ``/api/v4/control-changes/policies/{resource}/update``
Args:
resource : :obj:`str`, optional
Asset/Data Stream pair to update the associated Control Change Policy.
data: requests.ControlChangePolicyUpdate, optional
**kwargs:
Extra parameters for requests.ControlChangePolicyUpdate
- update_control_change_policy: str
"""
result_types = {
"200": responses.ControlChangePolicyUpdate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
"500": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/control-changes/policies/{resource}/update",
values={"resource": resource},
params={},
files={},
headers={},
data=data,
body_type=requests.ControlChangePolicyUpdate,
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.ControlChangePolicyUpdate,
)
@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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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,
nulls: Optional[Literal["first", "last"]] = 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 Control Changes created within a specified time range that match
the request filters.
**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')
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.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,
"nulls": nulls,
"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",
params=_request.params,
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
UUID of 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]:
"""Returns a Control Change by `control_change_id`, including its current
state and status history.
**Permission Required:** `kelvin.permission.control_change.read`.
``getControlChange``: ``GET`` ``/api/v4/control-changes/{control_change_id}/get``
Args:
control_change_id : :obj:`str`, optional
UUID of 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,
)