"""The endpoints for pending_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 .. import parsers, utils
if t.TYPE_CHECKING or os.getenv("CG_EAGERIMPORT", False):
from .. import client
from ..models.confirm_pending_registration_data import (
ConfirmPendingRegistrationData,
)
from ..models.confirm_quarantined_response import (
ConfirmQuarantinedResponse,
)
from ..models.confirm_registered_response import ConfirmRegisteredResponse
from ..models.create_pending_registration_data import (
CreatePendingRegistrationData,
)
from ..models.pending_registration_response import (
PendingRegistrationResponse,
)
_ClientT = t.TypeVar("_ClientT", bound="client._BaseClient")
[docs]
class PendingRegistrationService(t.Generic[_ClientT]):
__slots__ = ("__client",)
def __init__(self, client: _ClientT) -> None:
self.__client = client
[docs]
def confirm(
self,
json_body: ConfirmPendingRegistrationData,
*,
token: str,
) -> t.Union[ConfirmRegisteredResponse, ConfirmQuarantinedResponse]:
"""Verify the email code and finalize the registration.
The user provides the verification code along with their desired
username, name, and (optionally) password. If the username is already
taken, the code is not consumed and the user can retry.
:param json_body: The body of the request. See
:class:`.ConfirmPendingRegistrationData` for information about the
possible fields. You can provide this data as a
:class:`.ConfirmPendingRegistrationData` or as a dictionary.
:param token: The pending registration identifier.
:returns: A login response or quarantine status.
"""
url = "/api/v1/registrations/pending/{token}/confirm".format(
token=token
)
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.confirm_quarantined_response import (
ConfirmQuarantinedResponse,
)
from ..models.confirm_registered_response import (
ConfirmRegisteredResponse,
)
return parsers.JsonResponseParser(
parsers.make_union(
parsers.ParserFor.make(ConfirmRegisteredResponse),
parsers.ParserFor.make(ConfirmQuarantinedResponse),
)
).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 create(
self,
json_body: CreatePendingRegistrationData,
) -> PendingRegistrationResponse:
"""Create a pending registration and send a verification code.
Only an email address is required. The account details (username, name,
password) are provided when confirming the registration.
:param json_body: The body of the request. See
:class:`.CreatePendingRegistrationData` for information about the
possible fields. You can provide this data as a
:class:`.CreatePendingRegistrationData` or as a dictionary.
:returns: An identifier for the pending registration.
"""
url = "/api/v1/registrations/pending"
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.pending_registration_response import (
PendingRegistrationResponse,
)
return parsers.JsonResponseParser(
parsers.ParserFor.make(PendingRegistrationResponse)
).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),
),
),
)