# Code generated by builder. DO NOT EDIT.
"""
Kelvin API Client.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Optional, Union, overload
from typing_extensions import Literal
from kelvin.api.base.api_service_model import AsyncApiServiceModel
from kelvin.api.base.data_model import KList
from kelvin.api.base.http_client.base_client import AsyncBaseClient, Response
from ..model import requests, response, responses
[docs]
class UserAuthorization(AsyncApiServiceModel):
@overload
@classmethod
async def create_group(
cls,
data: Optional[Union[requests.GroupCreate, 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_group(
cls,
data: Optional[Union[requests.GroupCreate, 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_group(
cls,
data: Optional[Union[requests.GroupCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.GroupCreate: ...
[docs]
@classmethod
async def create_group(
cls,
data: Optional[Union[requests.GroupCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.GroupCreate, dict[str, object], Response]:
"""Creates a group
**Permission Required:** `kelvin.permission.authorization.create`.
``createGroup``: ``POST`` ``/api/v4/authorization/groups/create``
Args:
data: requests.GroupCreate, optional
**kwargs:
Extra parameters for requests.GroupCreate
- create_group: dict
"""
result_types = {
"200": responses.GroupCreate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
"409": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/authorization/groups/create",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.GroupCreate,
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.GroupCreate,
)
@overload
@classmethod
async def list_groups(
cls,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
role_names: 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,
fetch: bool = True,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def list_groups(
cls,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
role_names: 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,
fetch: bool = True,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def list_groups(
cls,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
role_names: 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,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.GroupsListPaginatedResponseCursor, responses.GroupsListPaginatedResponseLimits]: ...
@overload
@classmethod
async def list_groups(
cls,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
role_names: 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,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> KList[responses.GroupItem]: ...
[docs]
@classmethod
async def list_groups(
cls,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
role_names: 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,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
Union[
KList[responses.GroupItem],
responses.GroupsListPaginatedResponseCursor,
responses.GroupsListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a list of Groups and its parameters. The list can be optionally filtered and sorted on the server before being returned.
**Permission Required:** `kelvin.permission.authorization.read`.
``listGroups``: ``GET`` ``/api/v4/authorization/groups/list``
Args:
search : :obj:`Sequence[str]`
Search groups by key `name` or `title`. All values in array will be
filtered as `OR`. The search is case insensitive and will find partial
matches as well.
names : :obj:`Sequence[str]`
Filter Group list based on the key `name`.
role_names : :obj:`Sequence[str]`
Filter Group list based on the key `role_name`.
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')
"""
result_types = {
"200": responses.GroupsListPaginatedResponseCursor,
"400": response.Error,
"401": response.Error,
}
# override pagination_type
# stream type is only supported for raw responses
if not _get_response and pagination_type == "stream":
pagination_type = "cursor"
_request = cls._prepare_request(
method="GET",
path="/api/v4/authorization/groups/list",
values={},
params={
"search": search,
"names": names,
"role_names": role_names,
"sort_by": sort_by,
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
},
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.GroupsListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.GroupsListPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="GET",
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[responses.GroupItem],
)
return result
@overload
@classmethod
async def delete_group(
cls,
group_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_group(
cls,
group_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_group(
cls,
group_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_group(
cls,
group_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Delete a group.
**Permission Required:** `kelvin.permission.authorization.delete`.
``deleteGroup``: ``POST`` ``/api/v4/authorization/groups/{group_name}/delete``
Args:
group_name : :obj:`str`, optional
Fill the group name. The string can only contain lowercase
alphanumeric characters and `.`, `_` or `-` characters.
"""
result_types = {"200": None, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/authorization/groups/{group_name}/delete",
values={"group_name": group_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_group(
cls,
group_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_group(
cls,
group_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_group(
cls,
group_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.GroupGet: ...
[docs]
@classmethod
async def get_group(
cls,
group_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.GroupGet, dict[str, object], Response]:
"""Retrieve a Group.
**Permission Required:** `kelvin.permission.authorization.read`.
``getGroup``: ``GET`` ``/api/v4/authorization/groups/{group_name}/get``
Args:
group_name : :obj:`str`, optional
Fill the group name. The string can only contain lowercase
alphanumeric characters and `.`, `_` or `-` characters.
"""
result_types = {"200": responses.GroupGet, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="GET",
path="/api/v4/authorization/groups/{group_name}/get",
values={"group_name": group_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.GroupGet,
)
@overload
@classmethod
async def update_group(
cls,
group_name: str,
data: Optional[Union[requests.GroupUpdate, 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_group(
cls,
group_name: str,
data: Optional[Union[requests.GroupUpdate, 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_group(
cls,
group_name: str,
data: Optional[Union[requests.GroupUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.GroupUpdate: ...
[docs]
@classmethod
async def update_group(
cls,
group_name: str,
data: Optional[Union[requests.GroupUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.GroupUpdate, dict[str, object], Response]:
"""Update a group.
**Permission Required:** `kelvin.permission.authorization.update`.
``updateGroup``: ``POST`` ``/api/v4/authorization/groups/{group_name}/update``
Args:
group_name : :obj:`str`, optional
Fill the group name. The string can only contain lowercase
alphanumeric characters and `.`, `_` or `-` characters.
data: requests.GroupUpdate, optional
**kwargs:
Extra parameters for requests.GroupUpdate
- update_group: dict
"""
result_types = {
"200": responses.GroupUpdate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/authorization/groups/{group_name}/update",
values={"group_name": group_name},
params={},
files={},
headers={},
data=data,
body_type=requests.GroupUpdate,
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.GroupUpdate,
)
@overload
@classmethod
async def create_role(
cls,
data: Optional[Union[requests.RoleCreate, 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_role(
cls,
data: Optional[Union[requests.RoleCreate, 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_role(
cls,
data: Optional[Union[requests.RoleCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.RoleCreate: ...
[docs]
@classmethod
async def create_role(
cls,
data: Optional[Union[requests.RoleCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.RoleCreate, dict[str, object], Response]:
"""Creates a role
**Permission Required:** `kelvin.permission.authorization.create`.
``createRole``: ``POST`` ``/api/v4/authorization/roles/create``
Args:
data: requests.RoleCreate, optional
**kwargs:
Extra parameters for requests.RoleCreate
- create_role: dict
"""
result_types = {
"201": responses.RoleCreate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
"409": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/authorization/roles/create",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.RoleCreate,
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.RoleCreate,
)
@overload
@classmethod
async def list_roles(
cls,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
group_names: 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,
fetch: bool = True,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def list_roles(
cls,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
group_names: 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,
fetch: bool = True,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def list_roles(
cls,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
group_names: 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,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.RolesListPaginatedResponseCursor, responses.RolesListPaginatedResponseLimits]: ...
@overload
@classmethod
async def list_roles(
cls,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
group_names: 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,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> KList[responses.RoleItem]: ...
[docs]
@classmethod
async def list_roles(
cls,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
group_names: 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,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
Union[
KList[responses.RoleItem],
responses.RolesListPaginatedResponseCursor,
responses.RolesListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a list of Roles. The list can be optionally filtered and sorted on the server before being returned.
**Permission Required:** `kelvin.permission.authorization.read`.
``listRoles``: ``GET`` ``/api/v4/authorization/roles/list``
Args:
search : :obj:`Sequence[str]`
Search roles by key `name` or `title`. All values in array will be
filtered as `OR`. The search is case insensitive and will find partial
matches as well.
names : :obj:`Sequence[str]`
Filter Role list based on the key `name`.
group_names : :obj:`Sequence[str]`
Filter Role list based on the key `group_name`.
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')
"""
result_types = {"200": responses.RolesListPaginatedResponseCursor, "400": response.Error, "401": response.Error}
# override pagination_type
# stream type is only supported for raw responses
if not _get_response and pagination_type == "stream":
pagination_type = "cursor"
_request = cls._prepare_request(
method="GET",
path="/api/v4/authorization/roles/list",
values={},
params={
"search": search,
"names": names,
"group_names": group_names,
"sort_by": sort_by,
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
},
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.RolesListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.RolesListPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="GET",
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[responses.RoleItem],
)
return result
@overload
@classmethod
async def delete_role(
cls,
role_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_role(
cls,
role_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_role(
cls,
role_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_role(
cls,
role_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Delete a role.
**Permission Required:** `kelvin.permission.authorization.delete`.
``deleteRole``: ``POST`` ``/api/v4/authorization/roles/{role_name}/delete``
Args:
role_name : :obj:`str`, optional
Fill the role name. The string can only contain lowercase alphanumeric
characters and `.`, `_` or `-` characters.
"""
result_types = {"200": None, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/authorization/roles/{role_name}/delete",
values={"role_name": role_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_role(
cls,
role_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_role(
cls,
role_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_role(
cls,
role_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.RoleGet: ...
[docs]
@classmethod
async def get_role(
cls,
role_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.RoleGet, dict[str, object], Response]:
"""Retrieve a Role.
**Permission Required:** `kelvin.permission.authorization.read`.
``getRole``: ``GET`` ``/api/v4/authorization/roles/{role_name}/get``
Args:
role_name : :obj:`str`, optional
Fill the role name. The string can only contain lowercase alphanumeric
characters and `.`, `_` or `-` characters.
"""
result_types = {"200": responses.RoleGet, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="GET",
path="/api/v4/authorization/roles/{role_name}/get",
values={"role_name": role_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.RoleGet,
)
@overload
@classmethod
async def create_role_policy(
cls,
role_name: str,
data: Optional[Union[requests.RolePolicyCreate, 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_role_policy(
cls,
role_name: str,
data: Optional[Union[requests.RolePolicyCreate, 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_role_policy(
cls,
role_name: str,
data: Optional[Union[requests.RolePolicyCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.RolePolicyCreate: ...
[docs]
@classmethod
async def create_role_policy(
cls,
role_name: str,
data: Optional[Union[requests.RolePolicyCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.RolePolicyCreate, dict[str, object], Response]:
"""Creates a role policy
**Permission Required:** `kelvin.permission.authorization.create`.
``createRolePolicy``: ``POST`` ``/api/v4/authorization/roles/{role_name}/policies/create``
Args:
role_name : :obj:`str`, optional
Fill the role name. The string can only contain lowercase alphanumeric
characters and `.`, `_` or `-` characters.
data: requests.RolePolicyCreate, optional
**kwargs:
Extra parameters for requests.RolePolicyCreate
- create_role_policy: dict
"""
result_types = {
"201": responses.RolePolicyCreate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
"409": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/authorization/roles/{role_name}/policies/create",
values={"role_name": role_name},
params={},
files={},
headers={},
data=data,
body_type=requests.RolePolicyCreate,
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.RolePolicyCreate,
)
@overload
@classmethod
async def list_role_policies(
cls,
role_name: str,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
resource_types: 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,
fetch: bool = True,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def list_role_policies(
cls,
role_name: str,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
resource_types: 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,
fetch: bool = True,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def list_role_policies(
cls,
role_name: str,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
resource_types: 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,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
responses.RolePoliciesListPaginatedResponseCursor, responses.RolePoliciesListPaginatedResponseLimits
]: ...
@overload
@classmethod
async def list_role_policies(
cls,
role_name: str,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
resource_types: 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,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> KList[responses.RolePolicy]: ...
[docs]
@classmethod
async def list_role_policies(
cls,
role_name: str,
search: Optional[Sequence[str]] = None,
names: Optional[Sequence[str]] = None,
resource_types: 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,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[
Union[
KList[responses.RolePolicy],
responses.RolePoliciesListPaginatedResponseCursor,
responses.RolePoliciesListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a list of policies for a role. The list can be optionally filtered and sorted on the server before being returned.
**Permission Required:** `kelvin.permission.authorization.read`.
``listRolePolicies``: ``GET`` ``/api/v4/authorization/roles/{role_name}/policies/list``
Args:
role_name : :obj:`str`, optional
Fill the role name. The string can only contain lowercase alphanumeric
characters and `.`, `_` or `-` characters.
search : :obj:`Sequence[str]`
Search policies by key `name` or `title`. All values in array will be
filtered as `OR`. The search is case insensitive and will find partial
matches as well.
names : :obj:`Sequence[str]`
Filter policies list based on the key `name`.
resource_types : :obj:`Sequence[str]`
Filter policies list based on the key `resource_types`.
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')
"""
result_types = {
"200": responses.RolePoliciesListPaginatedResponseCursor,
"400": response.Error,
"401": response.Error,
}
# override pagination_type
# stream type is only supported for raw responses
if not _get_response and pagination_type == "stream":
pagination_type = "cursor"
_request = cls._prepare_request(
method="GET",
path="/api/v4/authorization/roles/{role_name}/policies/list",
values={"role_name": role_name},
params={
"search": search,
"names": names,
"resource_types": resource_types,
"sort_by": sort_by,
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
},
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.RolePoliciesListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.RolePoliciesListPaginatedResponseCursor,
)
if fetch:
return await cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="GET",
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[responses.RolePolicy],
)
return result
@overload
@classmethod
async def delete_role_policy(
cls,
role_name: str,
policy_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def delete_role_policy(
cls,
role_name: str,
policy_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def delete_role_policy(
cls,
role_name: str,
policy_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> None: ...
[docs]
@classmethod
async def delete_role_policy(
cls,
role_name: str,
policy_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Optional[Union[dict[str, object], Response]]:
"""Delete a role policy.
**Permission Required:** `kelvin.permission.authorization.delete`.
``deleteRolePolicy``: ``POST`` ``/api/v4/authorization/roles/{role_name}/policies/{policy_name}/delete``
Args:
role_name : :obj:`str`, optional
Fill the role name related to the policy. The string can only contain
lowercase alphanumeric characters and `.`, `_` or `-` characters.
policy_name : :obj:`str`, optional
Fill the policy name. The string can only contain lowercase
alphanumeric characters and `.`, `_` or `-` characters.
"""
result_types = {"200": None, "400": response.Error, "401": response.Error, "404": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/authorization/roles/{role_name}/policies/{policy_name}/delete",
values={"role_name": role_name, "policy_name": policy_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_role_policy(
cls,
role_name: str,
policy_name: str,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> dict[str, object]: ...
@overload
@classmethod
async def get_role_policy(
cls,
role_name: str,
policy_name: str,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[AsyncBaseClient] = None,
) -> Response: ...
@overload
@classmethod
async def get_role_policy(
cls,
role_name: str,
policy_name: str,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
) -> responses.RolePolicyGet: ...
[docs]
@classmethod
async def get_role_policy(
cls,
role_name: str,
policy_name: str,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
) -> Union[responses.RolePolicyGet, dict[str, object], Response]:
"""Retrieve a Role Policy.
**Permission Required:** `kelvin.permission.authorization.read`.
``getRolePolicy``: ``GET`` ``/api/v4/authorization/roles/{role_name}/policies/{policy_name}/get``
Args:
role_name : :obj:`str`, optional
Fill the role name related to the policy. The string can only contain
lowercase alphanumeric characters and `.`, `_` or `-` characters.
policy_name : :obj:`str`, optional
Fill the policy name. The string can only contain lowercase
alphanumeric characters and `.`, `_` or `-` characters.
"""
result_types = {
"200": responses.RolePolicyGet,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="GET",
path="/api/v4/authorization/roles/{role_name}/policies/{policy_name}/get",
values={"role_name": role_name, "policy_name": policy_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.RolePolicyGet,
)
@overload
@classmethod
async def update_role_policy(
cls,
role_name: str,
policy_name: str,
data: Optional[Union[requests.RolePolicyUpdate, 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_role_policy(
cls,
role_name: str,
policy_name: str,
data: Optional[Union[requests.RolePolicyUpdate, 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_role_policy(
cls,
role_name: str,
policy_name: str,
data: Optional[Union[requests.RolePolicyUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.RolePolicyUpdate: ...
[docs]
@classmethod
async def update_role_policy(
cls,
role_name: str,
policy_name: str,
data: Optional[Union[requests.RolePolicyUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.RolePolicyUpdate, dict[str, object], Response]:
"""Update a role policy.
**Permission Required:** `kelvin.permission.authorization.update`.
``updateRolePolicy``: ``POST`` ``/api/v4/authorization/roles/{role_name}/policies/{policy_name}/update``
Args:
role_name : :obj:`str`, optional
Fill the role name related to the policy. The string can only contain
lowercase alphanumeric characters and `.`, `_` or `-` characters.
policy_name : :obj:`str`, optional
Fill the policy name. The string can only contain lowercase
alphanumeric characters and `.`, `_` or `-` characters.
data: requests.RolePolicyUpdate, optional
**kwargs:
Extra parameters for requests.RolePolicyUpdate
- update_role_policy: dict
"""
result_types = {
"200": responses.RolePolicyUpdate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/authorization/roles/{role_name}/policies/{policy_name}/update",
values={"role_name": role_name, "policy_name": policy_name},
params={},
files={},
headers={},
data=data,
body_type=requests.RolePolicyUpdate,
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.RolePolicyUpdate,
)
@overload
@classmethod
async def update_role(
cls,
role_name: str,
data: Optional[Union[requests.RoleUpdate, 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_role(
cls,
role_name: str,
data: Optional[Union[requests.RoleUpdate, 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_role(
cls,
role_name: str,
data: Optional[Union[requests.RoleUpdate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> responses.RoleUpdate: ...
[docs]
@classmethod
async def update_role(
cls,
role_name: str,
data: Optional[Union[requests.RoleUpdate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[AsyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.RoleUpdate, dict[str, object], Response]:
"""Update a role.
**Permission Required:** `kelvin.permission.authorization.update`.
``updateRole``: ``POST`` ``/api/v4/authorization/roles/{role_name}/update``
Args:
role_name : :obj:`str`, optional
Fill the role name. The string can only contain lowercase alphanumeric
characters and `.`, `_` or `-` characters.
data: requests.RoleUpdate, optional
**kwargs:
Extra parameters for requests.RoleUpdate
- update_role: dict
"""
result_types = {
"200": responses.RoleUpdate,
"400": response.Error,
"401": response.Error,
"404": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/authorization/roles/{role_name}/update",
values={"role_name": role_name},
params={},
files={},
headers={},
data=data,
body_type=requests.RoleUpdate,
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.RoleUpdate,
)