"""The endpoints for group_set objects.
SPDX-License-Identifier: AGPL-3.0-only OR BSD-3-Clause-Clear
"""
from __future__ import annotations
import os
import typing as t
import cg_request_args as rqa
from cg_maybe import Maybe, Nothing
from cg_maybe.utils import maybe_from_nullable
from .. import paginated, parsers, utils
if t.TYPE_CHECKING or os.getenv("CG_EAGERIMPORT", False):
from .. import client
from ..models.create_group_group_set_data import CreateGroupGroupSetData
from ..models.extended_group import ExtendedGroup
from ..models.group import Group
from ..models.group_set import GroupSet
_ClientT = t.TypeVar("_ClientT", bound="client._BaseClient")
[docs]class GroupSetService(t.Generic[_ClientT]):
__slots__ = ("__client",)
def __init__(self, client: _ClientT) -> None:
self.__client = client
[docs] def create_group(
self: GroupSetService[client.AuthenticatedClient],
json_body: CreateGroupGroupSetData,
*,
group_set_id: int,
) -> ExtendedGroup:
"""Create a group for the given group set.
:param json_body: The body of the request. See
:class:`.CreateGroupGroupSetData` for information about the
possible fields. You can provide this data as a
:class:`.CreateGroupGroupSetData` or as a dictionary.
:param group_set_id: The id of the group set where the new group should
be placed in.
:returns: The newly created group.
"""
url = "/api/v1/group_sets/{groupSetId}/group".format(
groupSetId=group_set_id
)
params = None
with self.__client as client:
resp = client.http.post(
url=url, json=utils.to_dict(json_body), params=params
)
utils.log_warnings(resp)
if utils.response_code_matches(resp.status_code, 200):
from ..models.extended_group import ExtendedGroup
return parsers.JsonResponseParser(
parsers.ParserFor.make(ExtendedGroup)
).try_parse(resp)
from ..models.any_error import AnyError
raise utils.get_error(
resp,
(
(
(400, 409, 401, 403, 404, 429, 500),
utils.unpack_union(AnyError),
),
),
)
[docs] def get(
self: GroupSetService[client.AuthenticatedClient],
*,
group_set_id: int,
) -> GroupSet:
"""Get a single `GroupSet` by id.
:param group_set_id: The id of the group set
:returns: A response containing the JSON serialized group set.
"""
url = "/api/v1/group_sets/{groupSetId}".format(groupSetId=group_set_id)
params = None
with self.__client as client:
resp = client.http.get(url=url, params=params)
utils.log_warnings(resp)
if utils.response_code_matches(resp.status_code, 200):
from ..models.group_set import GroupSet
return parsers.JsonResponseParser(
parsers.ParserFor.make(GroupSet)
).try_parse(resp)
from ..models.any_error import AnyError
raise utils.get_error(
resp,
(
(
(400, 409, 401, 403, 404, 429, 500),
utils.unpack_union(AnyError),
),
),
)
[docs] def delete(
self: GroupSetService[client.AuthenticatedClient],
*,
group_set_id: int,
) -> None:
"""Delete a `GroupSet`.
You can only delete a group set if there are no groups in the set and
no assignment is connected to the group set.
:param group_set_id: The id of the group set
:returns: Nothing.
"""
url = "/api/v1/group_sets/{groupSetId}".format(groupSetId=group_set_id)
params = None
with self.__client as client:
resp = client.http.delete(url=url, params=params)
utils.log_warnings(resp)
if utils.response_code_matches(resp.status_code, 204):
return parsers.ConstantlyParser(None).try_parse(resp)
from ..models.any_error import AnyError
raise utils.get_error(
resp,
(
(
(400, 409, 401, 403, 404, 429, 500),
utils.unpack_union(AnyError),
),
),
)
[docs] def get_all_groups(
self: GroupSetService[client.AuthenticatedClient],
*,
group_set_id: int,
page_size: int = 20,
) -> paginated.Response[Group]:
"""Get all groups for a given group set.
:param group_set_id: The group set for which the groups should be
returned.
:param page_size: The size of a single page, maximum is 50.
:returns: All the groups for the given group set.
"""
url = "/api/v1/group_sets/{groupSetId}/groups/".format(
groupSetId=group_set_id
)
params: t.Dict[str, str | int | bool] = {
"page-size": page_size,
}
if t.TYPE_CHECKING:
import httpx
def do_request(next_token: str | None) -> httpx.Response:
if next_token is None:
params.pop("next-token", "")
else:
params["next-token"] = next_token
with self.__client as client:
resp = client.http.get(url=url, params=params)
utils.log_warnings(resp)
return resp
def parse_response(resp: httpx.Response) -> t.Sequence[Group]:
if utils.response_code_matches(resp.status_code, 200):
from ..models.group import Group
return parsers.JsonResponseParser(
rqa.List(parsers.ParserFor.make(Group))
).try_parse(resp)
from ..models.any_error import AnyError
raise utils.get_error(
resp,
(
(
(400, 409, 401, 403, 404, 429, 500),
utils.unpack_union(AnyError),
),
),
)
return paginated.Response(do_request, parse_response)
[docs] def get_user_group(
self: GroupSetService[client.AuthenticatedClient],
*,
group_set_id: int,
user_id: int,
) -> ExtendedGroup:
"""Get the group for a specific user within a given group set.
:param group_set_id: The group set to search within.
:param user_id: The ID of the user whose group is to be found.
:returns: The single group object for the given user.
"""
url = "/api/v1/group_sets/{groupSetId}/user/{userId}/group".format(
groupSetId=group_set_id, userId=user_id
)
params = None
with self.__client as client:
resp = client.http.get(url=url, params=params)
utils.log_warnings(resp)
if utils.response_code_matches(resp.status_code, 200):
from ..models.extended_group import ExtendedGroup
return parsers.JsonResponseParser(
parsers.ParserFor.make(ExtendedGroup)
).try_parse(resp)
from ..models.any_error import AnyError
raise utils.get_error(
resp,
(
(
(400, 409, 401, 403, 404, 429, 500),
utils.unpack_union(AnyError),
),
),
)