"""The endpoints for quarantined_registration 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.accept_quarantined_registration_data import (
AcceptQuarantinedRegistrationData,
)
from ..models.archive_quarantined_registration_data import (
ArchiveQuarantinedRegistrationData,
)
from ..models.deny_quarantined_registration_data import (
DenyQuarantinedRegistrationData,
)
from ..models.quarantined_registration import QuarantinedRegistration
_ClientT = t.TypeVar("_ClientT", bound="client._BaseClient")
[docs]
class QuarantinedRegistrationService(t.Generic[_ClientT]):
__slots__ = ("__client",)
def __init__(self, client: _ClientT) -> None:
self.__client = client
[docs]
def accept(
self: QuarantinedRegistrationService[client.AuthenticatedClient],
json_body: AcceptQuarantinedRegistrationData,
*,
quarantined_registration_id: str,
) -> None:
"""Accept quarantined registrations.
:param json_body: The body of the request. See
:class:`.AcceptQuarantinedRegistrationData` for information about
the possible fields. You can provide this data as a
:class:`.AcceptQuarantinedRegistrationData` or as a dictionary.
:param quarantined_registration_id: Registration id to accept.
:returns: An empty response with return code 204
"""
url = "/api/v1/quarantined_registrations/{quarantinedRegistrationId}/accept".format(
quarantinedRegistrationId=quarantined_registration_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, 204):
return parsers.ConstantlyParser(None).try_parse(resp)
from ..models.any_error import AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)
[docs]
def archive(
self: QuarantinedRegistrationService[client.AuthenticatedClient],
json_body: ArchiveQuarantinedRegistrationData,
) -> None:
"""Archive all active quarantined registrations for an email address.
This clears duplicate signups from the board without notifying the
user. Every active registration for the email is archived, not just a
single row, because duplicates may span multiple pages of the list.
:param json_body: The body of the request. See
:class:`.ArchiveQuarantinedRegistrationData` for information about
the possible fields. You can provide this data as a
:class:`.ArchiveQuarantinedRegistrationData` or as a dictionary.
:returns: An empty response with return code 204
"""
url = "/api/v1/quarantined_registrations/archive"
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, 204):
return parsers.ConstantlyParser(None).try_parse(resp)
from ..models.any_error import AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)
[docs]
def deny(
self: QuarantinedRegistrationService[client.AuthenticatedClient],
json_body: DenyQuarantinedRegistrationData,
*,
quarantined_registration_id: str,
) -> None:
"""Deny quarantined registrations.
:param json_body: The body of the request. See
:class:`.DenyQuarantinedRegistrationData` for information about the
possible fields. You can provide this data as a
:class:`.DenyQuarantinedRegistrationData` or as a dictionary.
:param quarantined_registration_id: Registration id to accept.
:returns: An empty response with return code 204
"""
url = "/api/v1/quarantined_registrations/{quarantinedRegistrationId}/deny".format(
quarantinedRegistrationId=quarantined_registration_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, 204):
return parsers.ConstantlyParser(None).try_parse(resp)
from ..models.any_error import AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)
[docs]
def get_all(
self: QuarantinedRegistrationService[client.AuthenticatedClient],
*,
is_archived: bool = False,
page_size: int = 25,
) -> paginated.Response[QuarantinedRegistration]:
"""List quarantined registrations.
:param is_archived: If `True` return archived registrations, otherwise
return the active ones awaiting review.
:param page_size: The size of a single page, maximum is 100.
:returns: Paginated list of quarantined registrations.
"""
url = "/api/v1/quarantined_registrations/"
params: t.Dict[str, str | int | bool] = {
"is_archived": is_archived,
"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[QuarantinedRegistration]:
if utils.response_code_matches(resp.status_code, 200):
from ..models.quarantined_registration import (
QuarantinedRegistration,
)
return parsers.JsonResponseParser(
rqa.List(parsers.ParserFor.make(QuarantinedRegistration))
).try_parse(resp)
from ..models.any_error import AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)
return paginated.Response(do_request, parse_response)
[docs]
def unarchive(
self: QuarantinedRegistrationService[client.AuthenticatedClient],
*,
quarantined_registration_id: str,
) -> None:
"""Restore an archived quarantined registration to the active board.
:param quarantined_registration_id: Registration id to unarchive.
:returns: An empty response with return code 204
"""
url = "/api/v1/quarantined_registrations/{quarantinedRegistrationId}/unarchive".format(
quarantinedRegistrationId=quarantined_registration_id
)
params = None
with self.__client as client:
resp = client.http.post(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 AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)