# 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 Apps(AsyncApiServiceModel):
@overload
@classmethod
async def list_apps_context(
cls,
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,
data: Optional[Union[requests.AppsContextList, 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_apps_context(
cls,
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,
data: Optional[Union[requests.AppsContextList, 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_apps_context(
cls,
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,
data: Optional[Union[requests.AppsContextList, 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.AppsContextListPaginatedResponseCursor, responses.AppsContextListPaginatedResponseLimits]: ...
@overload
@classmethod
async def list_apps_context(
cls,
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,
data: Optional[Union[requests.AppsContextList, 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.AppsResourceContext]: ...
[docs]
@classmethod
async def list_apps_context(
cls,
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,
data: Optional[Union[requests.AppsContextList, 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.AppsResourceContext],
responses.AppsContextListPaginatedResponseCursor,
responses.AppsContextListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns App resource contexts that match the request filters.
**Permission Required:** `kelvin.permission.app.read`.
``listAppsContext``: ``POST`` ``/api/v4/apps/context/list``
Args:
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')
data: requests.AppsContextList, optional
**kwargs:
Extra parameters for requests.AppsContextList
- list_apps_context: dict
"""
result_types = {
"200": responses.AppsContextListPaginatedResponseCursor,
"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/apps/context/list",
values={},
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=data,
body_type=requests.AppsContextList,
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.AppsContextListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.AppsContextListPaginatedResponseCursor,
)
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.AppsResourceContext],
)
return result
@overload
@classmethod
async def create_app_version(
cls,
data: Optional[Union[requests.AppVersionCreate, 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_app_version(
cls,
data: Optional[Union[requests.AppVersionCreate, 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_app_version(
cls,
data: Optional[Union[requests.AppVersionCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.AppVersionCreate: ...
[docs]
@classmethod
async def create_app_version(
cls,
data: Optional[Union[requests.AppVersionCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.AppVersionCreate, dict[str, object], Response]:
"""Create App Version
**Permission Required:** `kelvin.permission.app.create`.
``createAppVersion``: ``POST`` ``/api/v4/apps/create``
Args:
data: requests.AppVersionCreate, optional
**kwargs:
Extra parameters for requests.AppVersionCreate
- create_app_version: str
"""
result_types = {
"201": responses.AppVersionCreate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/apps/create",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.AppVersionCreate,
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.AppVersionCreate,
)
@overload
@classmethod
async def list_apps(
cls,
app_names: Optional[Sequence[str]] = None,
app_types: Optional[Sequence[str]] = None,
resources: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
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_apps(
cls,
app_names: Optional[Sequence[str]] = None,
app_types: Optional[Sequence[str]] = None,
resources: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
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_apps(
cls,
app_names: Optional[Sequence[str]] = None,
app_types: Optional[Sequence[str]] = None,
resources: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
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.AppsListPaginatedResponseCursor, responses.AppsListPaginatedResponseLimits]: ...
@overload
@classmethod
async def list_apps(
cls,
app_names: Optional[Sequence[str]] = None,
app_types: Optional[Sequence[str]] = None,
resources: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
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.AppShort]: ...
[docs]
@classmethod
async def list_apps(
cls,
app_names: Optional[Sequence[str]] = None,
app_types: Optional[Sequence[str]] = None,
resources: Optional[Sequence[str]] = None,
search: Optional[Sequence[str]] = None,
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.AppShort], responses.AppsListPaginatedResponseCursor, responses.AppsListPaginatedResponseLimits
],
dict[str, object],
Response,
]:
"""Returns Apps that match the provided filters.
**Permission Required:** `kelvin.permission.app.read`.
``listApps``: ``GET`` ``/api/v4/apps/list``
Args:
app_names : :obj:`Sequence[str]`
Return only Apps whose `name` matches one or more values.
app_types : :obj:`Sequence[str]`
Return only Apps whose `type` matches one or more values.
resources : :obj:`Sequence[str]`
Return only Apps associated with the provided resource KRNs. Allowed
namespace is `asset`.
search : :obj:`Sequence[str]`
Case-insensitive partial match against App `name` and `title`.
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.AppsListPaginatedResponseCursor,
"400": response.Error,
"401": response.Error,
"404": 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/apps/list",
values={},
params={
"app_names": app_names,
"app_types": app_types,
"resources": resources,
"search": search,
"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.AppsListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.AppsListPaginatedResponseCursor,
)
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.AppShort],
)
return result
@overload
@classmethod
async def delete_app(
cls,
app_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_app(
cls,
app_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_app(
cls,
app_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_app(
cls,
app_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Delete App
**Permission Required:** `kelvin.permission.app.delete`.
``deleteApp``: ``POST`` ``/api/v4/apps/{app_name}/delete``
Args:
app_name : :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/apps/{app_name}/delete",
values={"app_name": app_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_app(
cls,
app_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_app(
cls,
app_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_app(
cls,
app_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.AppGet: ...
[docs]
@classmethod
async def get_app(
cls,
app_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.AppGet, dict[str, object], Response]:
"""Returns App details for the provided `app_name`.
**Permission Required:** `kelvin.permission.app.read`.
``getApp``: ``GET`` ``/api/v4/apps/{app_name}/get``
Args:
app_name : :obj:`str`, optional
Unique App name.
"""
result_types = {"200": responses.AppGet, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="GET",
path="/api/v4/apps/{app_name}/get",
values={"app_name": app_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.AppGet,
)
@overload
@classmethod
async def patch_app(
cls,
app_name: str,
data: Optional[Union[requests.AppPatch, 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 patch_app(
cls,
app_name: str,
data: Optional[Union[requests.AppPatch, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def patch_app(
cls,
app_name: str,
data: Optional[Union[requests.AppPatch, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.AppPatch: ...
[docs]
@classmethod
async def patch_app(
cls,
app_name: str,
data: Optional[Union[requests.AppPatch, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.AppPatch, dict[str, object], Response]:
"""Update details of an Application. Any parameters that are not provided will
remain unchanged.
**Permission Required:** `kelvin.permission.app.update`.
``patchApp``: ``POST`` ``/api/v4/apps/{app_name}/patch``
Args:
app_name : :obj:`str`, optional
data: requests.AppPatch, optional
**kwargs:
Extra parameters for requests.AppPatch
- patch_app: dict
"""
result_types = {"200": responses.AppPatch, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/apps/{app_name}/patch",
values={"app_name": app_name},
params={},
files={},
headers={},
data=data,
body_type=requests.AppPatch,
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.AppPatch,
)
@overload
@classmethod
async def delete_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesDelete, 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 delete_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesDelete, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def delete_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesDelete, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> None: ...
[docs]
@classmethod
async def delete_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesDelete, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Optional[Union[dict[str, object], Response]]:
"""Delete App Resources
**Permission Required:** `kelvin.permission.app.update`.
``deleteAppResources``: ``POST`` ``/api/v4/apps/{app_name}/resources/delete``
Args:
app_name : :obj:`str`, optional
data: requests.AppResourcesDelete, optional
**kwargs:
Extra parameters for requests.AppResourcesDelete
- delete_app_resources: dict
"""
result_types = {"204": None, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/apps/{app_name}/resources/delete",
values={"app_name": app_name},
params={},
files={},
headers={},
data=data,
body_type=requests.AppResourcesDelete,
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,
)
@overload
@classmethod
async def disable_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesDisable, 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 disable_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesDisable, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def disable_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesDisable, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> None: ...
[docs]
@classmethod
async def disable_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesDisable, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Optional[Union[dict[str, object], Response]]:
"""Disable App Resources
**Permission Required:** `kelvin.permission.app.update`.
``disableAppResources``: ``POST`` ``/api/v4/apps/{app_name}/resources/disable``
Args:
app_name : :obj:`str`, optional
data: requests.AppResourcesDisable, optional
**kwargs:
Extra parameters for requests.AppResourcesDisable
- disable_app_resources: dict
"""
result_types = {"204": None, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/apps/{app_name}/resources/disable",
values={"app_name": app_name},
params={},
files={},
headers={},
data=data,
body_type=requests.AppResourcesDisable,
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,
)
@overload
@classmethod
async def enable_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesEnable, 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 enable_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesEnable, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def enable_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesEnable, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> None: ...
[docs]
@classmethod
async def enable_app_resources(
cls,
app_name: str,
data: Optional[Union[requests.AppResourcesEnable, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Optional[Union[dict[str, object], Response]]:
"""Enable App Resources
**Permission Required:** `kelvin.permission.app.update`.
``enableAppResources``: ``POST`` ``/api/v4/apps/{app_name}/resources/enable``
Args:
app_name : :obj:`str`, optional
data: requests.AppResourcesEnable, optional
**kwargs:
Extra parameters for requests.AppResourcesEnable
- enable_app_resources: dict
"""
result_types = {"204": None, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/apps/{app_name}/resources/enable",
values={"app_name": app_name},
params={},
files={},
headers={},
data=data,
body_type=requests.AppResourcesEnable,
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,
)
@overload
@classmethod
async def list_app_resources(
cls,
app_name: str,
search: Optional[Sequence[str]] = None,
resources: Optional[Sequence[str]] = None,
app_versions: Optional[Sequence[str]] = None,
workload_names: Optional[Sequence[str]] = None,
enabled: Optional[bool] = None,
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_app_resources(
cls,
app_name: str,
search: Optional[Sequence[str]] = None,
resources: Optional[Sequence[str]] = None,
app_versions: Optional[Sequence[str]] = None,
workload_names: Optional[Sequence[str]] = None,
enabled: Optional[bool] = None,
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_app_resources(
cls,
app_name: str,
search: Optional[Sequence[str]] = None,
resources: Optional[Sequence[str]] = None,
app_versions: Optional[Sequence[str]] = None,
workload_names: Optional[Sequence[str]] = None,
enabled: Optional[bool] = None,
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.AppResourcesListPaginatedResponseCursor, responses.AppResourcesListPaginatedResponseLimits
]: ...
@overload
@classmethod
async def list_app_resources(
cls,
app_name: str,
search: Optional[Sequence[str]] = None,
resources: Optional[Sequence[str]] = None,
app_versions: Optional[Sequence[str]] = None,
workload_names: Optional[Sequence[str]] = None,
enabled: Optional[bool] = None,
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.AppResource]: ...
[docs]
@classmethod
async def list_app_resources(
cls,
app_name: str,
search: Optional[Sequence[str]] = None,
resources: Optional[Sequence[str]] = None,
app_versions: Optional[Sequence[str]] = None,
workload_names: Optional[Sequence[str]] = None,
enabled: Optional[bool] = None,
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.AppResource],
responses.AppResourcesListPaginatedResponseCursor,
responses.AppResourcesListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""List App Resources
**Permission Required:** `kelvin.permission.app.read`.
``listAppResources``: ``GET`` ``/api/v4/apps/{app_name}/resources/list``
Args:
app_name : :obj:`str`, optional
search : :obj:`Sequence[str]`
resources : :obj:`Sequence[str]`
app_versions : :obj:`Sequence[str]`
workload_names : :obj:`Sequence[str]`
enabled : :obj:`bool`
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.AppResourcesListPaginatedResponseCursor,
"400": response.Error,
"401": response.Error,
"404": 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/apps/{app_name}/resources/list",
values={"app_name": app_name},
params={
"search": search,
"resources": resources,
"app_versions": app_versions,
"workload_names": workload_names,
"enabled": enabled,
"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.AppResourcesListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.AppResourcesListPaginatedResponseCursor,
)
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.AppResource],
)
return result
@overload
@classmethod
async def delete_app_version(
cls,
app_name: str,
app_version: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_app_version(
cls,
app_name: str,
app_version: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_app_version(
cls,
app_name: str,
app_version: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_app_version(
cls,
app_name: str,
app_version: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Delete App Version
**Permission Required:** `kelvin.permission.app.delete`.
``deleteAppVersion``: ``POST`` ``/api/v4/apps/{app_name}/v/{app_version}/delete``
Args:
app_name : :obj:`str`, optional
app_version : :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/apps/{app_name}/v/{app_version}/delete",
values={"app_name": app_name, "app_version": app_version},
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 deploy_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionDeploy, 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 deploy_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionDeploy, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def deploy_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionDeploy, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.AppVersionDeploy: ...
[docs]
@classmethod
async def deploy_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionDeploy, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.AppVersionDeploy, dict[str, object], Response]:
"""Deploy App Version
Issues the deployment of workloads based on the specified instructions.
The `deployment` section defines the strategy to deploy the application to
resources (i.e. assets) by creating the necessary workloads based on the
maximum number of resources each instance can handle, as well as the target
cluster where the workloads will be deployed.
Each resource (i.e. asset) is defined in the `runtime` section, which
includes defining its parameters and data stream mappings.
The `parameters` section is optional. If provided, it will update all
parameters for that resource with the provided values, otherwise, the
current ones will be injected into the new workloads. If `parameters` is
set and a parameter is not defined, the current value will be deleted,
effectively setting it to the default value. This behaviour means that
setting `parameters` to an empty object (`{}`) will reset all parameters of
that resource to their default values.
**Permission Required:** `kelvin.permission.app.deploy`.
``deployAppVersion``: ``POST`` ``/api/v4/apps/{app_name}/v/{app_version}/deploy``
Args:
app_name : :obj:`str`, optional
app_version : :obj:`str`, optional
data: requests.AppVersionDeploy, optional
**kwargs:
Extra parameters for requests.AppVersionDeploy
- deploy_app_version: dict
"""
result_types = {
"200": responses.AppVersionDeploy,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/apps/{app_name}/v/{app_version}/deploy",
values={"app_name": app_name, "app_version": app_version},
params={},
files={},
headers={},
data=data,
body_type=requests.AppVersionDeploy,
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.AppVersionDeploy,
)
@overload
@classmethod
async def get_app_version(
cls,
app_name: str,
app_version: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_app_version(
cls,
app_name: str,
app_version: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_app_version(
cls,
app_name: str,
app_version: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.AppVersionGet: ...
[docs]
@classmethod
async def get_app_version(
cls,
app_name: str,
app_version: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.AppVersionGet, dict[str, object], Response]:
"""Returns details for the specified App version.
**Permission Required:** `kelvin.permission.app.read`.
``getAppVersion``: ``GET`` ``/api/v4/apps/{app_name}/v/{app_version}/get``
Args:
app_name : :obj:`str`, optional
Unique App name.
app_version : :obj:`str`, optional
Version identifier of the App.
"""
result_types = {
"200": responses.AppVersionGet,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="GET",
path="/api/v4/apps/{app_name}/v/{app_version}/get",
values={"app_name": app_name, "app_version": app_version},
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.AppVersionGet,
)
@overload
@classmethod
async def patch_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionPatch, 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 patch_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionPatch, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
async def patch_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionPatch, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.AppVersionPatch: ...
[docs]
@classmethod
async def patch_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionPatch, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.AppVersionPatch, dict[str, object], Response]:
"""Update App Version
Partially update the default settings and schemas for an App Version. The
following keys can be updated individually. If a key is not specified, its
existing value will remain unchanged. If a key is specified, it will
completely overwrite the current value for that key and its nested fields.
- `image`
- `defaults.deployment.max_resources`
- `defaults.deployment.deployment_type`
- `defaults.deployment.target`
- `defaults.app.configuration`
- `defaults.app.io_datastream_mapping`
- `defaults.api_permissions`
- `defaults.system`
- `schemas.parameters`
- `schemas.configuration`
- `schemas.io_configuration`
**Permission Required:** `kelvin.permission.app.update`.
``patchAppVersion``: ``POST`` ``/api/v4/apps/{app_name}/v/{app_version}/patch``
Args:
app_name : :obj:`str`, optional
app_version : :obj:`str`, optional
data: requests.AppVersionPatch, optional
**kwargs:
Extra parameters for requests.AppVersionPatch
- patch_app_version: dict
"""
result_types = {
"200": responses.AppVersionPatch,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/apps/{app_name}/v/{app_version}/patch",
values={"app_name": app_name, "app_version": app_version},
params={},
files={},
headers={},
data=data,
body_type=requests.AppVersionPatch,
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.AppVersionPatch,
)
@overload
@classmethod
async def update_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionUpdate, 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_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionUpdate, 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_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.AppVersionUpdate: ...
[docs]
@classmethod
async def update_app_version(
cls,
app_name: str,
app_version: str,
data: Optional[Union[requests.AppVersionUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.AppVersionUpdate, dict[str, object], Response]:
"""Update App Version
Update the default settings and schemas for an App Version.
**Permission Required:** `kelvin.permission.app.update`.
``updateAppVersion``: ``POST`` ``/api/v4/apps/{app_name}/v/{app_version}/update``
Args:
app_name : :obj:`str`, optional
app_version : :obj:`str`, optional
data: requests.AppVersionUpdate, optional
**kwargs:
Extra parameters for requests.AppVersionUpdate
- update_app_version: dict
"""
result_types = {
"200": responses.AppVersionUpdate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/apps/{app_name}/v/{app_version}/update",
values={"app_name": app_name, "app_version": app_version},
params={},
files={},
headers={},
data=data,
body_type=requests.AppVersionUpdate,
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.AppVersionUpdate,
)