"""The endpoints for restriction 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_entry_ticket_restriction_data import (
CreateEntryTicketRestrictionData,
)
from ..models.direct_entry_ticket_response import DirectEntryTicketResponse
from ..models.enter_restriction_data import EnterRestrictionData
from ..models.entry_overview_entry import EntryOverviewEntry
from ..models.exam_environment_response import ExamEnvironmentResponse
from ..models.ip_range import IPRange
from ..models.kiosk_launch import KioskLaunch
from ..models.launch_exam_environment_restriction_data import (
LaunchExamEnvironmentRestrictionData,
)
from ..models.non_direct_entry_ticket_response import (
NonDirectEntryTicketResponse,
)
from ..models.patch_entry_override_restriction_data import (
PatchEntryOverrideRestrictionData,
)
from ..models.patch_restriction_data import PatchRestrictionData
from ..models.refreshed_heartbeat_response import (
RefreshedHeartbeatResponse,
)
from ..models.restriction import Restriction
from ..models.restriction_with_effective_window import (
RestrictionWithEffectiveWindow,
)
from ..models.teacher_ui_url_response import TeacherUIUrlResponse
from ..models.user_login_response import UserLoginResponse
from ..models.verification_failed_heartbeat_response import (
VerificationFailedHeartbeatResponse,
)
from ..models.verification_unavailable_heartbeat_response import (
VerificationUnavailableHeartbeatResponse,
)
_ClientT = t.TypeVar("_ClientT", bound="client._BaseClient")
[docs]
class RestrictionService(t.Generic[_ClientT]):
__slots__ = ("__client",)
def __init__(self, client: _ClientT) -> None:
self.__client = client
[docs]
def create_entry_ticket(
self: RestrictionService[client.AuthenticatedClient],
json_body: CreateEntryTicketRestrictionData,
*,
restriction_id: str,
) -> t.Union[DirectEntryTicketResponse, NonDirectEntryTicketResponse]:
"""Create a short-lived entry ticket.
Authenticates the caller via bearer token, validates the password (when
set), and pre-checks the entry limit so a dead ticket isn't issued.
Writes an unused ticket to Redis and returns the ticket string.
Rejects when the calling token is itself an exam token for this course,
since activating another ticket would kill the active session.
:param json_body: The body of the request. See
:class:`.CreateEntryTicketRestrictionData` for information about
the possible fields. You can provide this data as a
:class:`.CreateEntryTicketRestrictionData` or as a dictionary.
:param restriction_id: The id of the restriction.
:returns: A tagged response (EntryTicketResponse) carrying the ticket.
"""
url = "/api/v1/restrictions/{restrictionId}/entry_tickets/".format(
restrictionId=restriction_id
)
params = None
with self.__client as client:
resp = client.http.put(
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.direct_entry_ticket_response import (
DirectEntryTicketResponse,
)
from ..models.non_direct_entry_ticket_response import (
NonDirectEntryTicketResponse,
)
return parsers.JsonResponseParser(
parsers.make_union(
parsers.ParserFor.make(DirectEntryTicketResponse),
parsers.ParserFor.make(NonDirectEntryTicketResponse),
)
).try_parse(resp)
from ..models.any_error import AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)
[docs]
def put_allowed_ip(
self: RestrictionService[client.AuthenticatedClient],
*,
restriction_id: str,
ip_range: str,
) -> IPRange:
"""Add a single IP range to a restriction.
:param restriction_id: The id of the restriction.
:param ip_range: The IP range to add in CIDR notation.
:returns: The added IP range.
"""
url = "/api/v1/restrictions/{restrictionId}/allowed_ips/{ipRange}".format(
restrictionId=restriction_id, ipRange=ip_range
)
params = None
with self.__client as client:
resp = client.http.put(url=url, params=params)
utils.log_warnings(resp)
if utils.response_code_matches(resp.status_code, 200):
from ..models.ip_range import IPRange
return parsers.JsonResponseParser(
parsers.ParserFor.make(IPRange)
).try_parse(resp)
from ..models.any_error import AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)
[docs]
def delete_allowed_ip(
self: RestrictionService[client.AuthenticatedClient],
*,
restriction_id: str,
ip_range: str,
) -> None:
"""Remove a single IP range from a restriction.
:param restriction_id: The ID of the restriction.
:param ip_range: The IP range to remove.
:returns: An empty response.
"""
url = "/api/v1/restrictions/{restrictionId}/allowed_ips/{ipRange}".format(
restrictionId=restriction_id, ipRange=ip_range
)
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 AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)
[docs]
def get_all_entries(
self: RestrictionService[client.AuthenticatedClient],
*,
restriction_id: str,
q: str = "",
page_size: int = 50,
) -> paginated.Response[EntryOverviewEntry]:
"""List all enrolled students with their entry counts and overrides.
:param restriction_id: The ID of the restriction.
:param q: Only retrieve students whose name or username matches this
value.
:param page_size: The size of a single page, maximum is 100.
:returns: A paginated list of entry overview entries per student.
"""
url = "/api/v1/restrictions/{restrictionId}/entries/".format(
restrictionId=restriction_id
)
params: t.Dict[str, str | int | bool] = {
"q": q,
"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[EntryOverviewEntry]:
if utils.response_code_matches(resp.status_code, 200):
from ..models.entry_overview_entry import EntryOverviewEntry
return parsers.JsonResponseParser(
rqa.List(parsers.ParserFor.make(EntryOverviewEntry))
).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 enter(
self,
json_body: EnterRestrictionData,
*,
restriction_id: str,
) -> UserLoginResponse:
"""Activate a previously created entry ticket.
The ticket is the credential, the `Authorization` header is ignored,
and a new session token is created while every previously issued token
for the user is invalidated.
:param json_body: The body of the request. See
:class:`.EnterRestrictionData` for information about the possible
fields. You can provide this data as a
:class:`.EnterRestrictionData` or as a dictionary.
:param restriction_id: The ID of the restriction.
:returns: A login response containing the new access token.
"""
url = "/api/v1/restrictions/{restrictionId}/entries/".format(
restrictionId=restriction_id
)
params = None
with self.__client as client:
resp = client.http.put(
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.user_login_response import UserLoginResponse
return parsers.JsonResponseParser(
parsers.ParserFor.make(UserLoginResponse)
).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_allowed_ips(
self: RestrictionService[client.AuthenticatedClient],
*,
restriction_id: str,
page_size: int = 50,
) -> paginated.Response[IPRange]:
"""Get all IP ranges for a restriction.
:param restriction_id: The ID of the restriction.
:param page_size: The size of a single page, maximum is 100.
:returns: A paginated list of IP ranges.
"""
url = "/api/v1/restrictions/{restrictionId}/allowed_ips/".format(
restrictionId=restriction_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[IPRange]:
if utils.response_code_matches(resp.status_code, 200):
from ..models.ip_range import IPRange
return parsers.JsonResponseParser(
rqa.List(parsers.ParserFor.make(IPRange))
).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 get_exam_environment(
self: RestrictionService[client.AuthenticatedClient],
*,
restriction_id: str,
) -> ExamEnvironmentResponse:
"""Get the exam environment attachment options for a restriction.
Unlike the other exam environment routes this never fails on the
release flag: with the feature off it reports no available providers,
so clients drive their management UI off this list alone.
:param restriction_id: The id of the restriction.
:returns: The providers the tenant can attach.
"""
url = "/api/v1/restrictions/{restrictionId}/exam_environment/".format(
restrictionId=restriction_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.exam_environment_response import (
ExamEnvironmentResponse,
)
return parsers.JsonResponseParser(
parsers.ParserFor.make(ExamEnvironmentResponse)
).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_exam_environment_teacher_ui(
self: RestrictionService[client.AuthenticatedClient],
*,
restriction_id: str,
kind: str,
) -> TeacherUIUrlResponse:
"""Get a fresh deep link into the provider-hosted teacher UI.
`kind` selects the UI: `settings` is the provider's exam-configuration
widget, `dashboard` its live proctoring dashboard. Every call fetches a
fresh single-use URL from the provider, so clients must request one per
open and never store it.
:param restriction_id: The id of the restriction.
:param kind: The teacher UI to deep-link into.
:returns: The single-use URL to open.
"""
url = "/api/v1/restrictions/{restrictionId}/exam_environment/teacher_ui/{kind}".format(
restrictionId=restriction_id, kind=kind
)
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.teacher_ui_url_response import TeacherUIUrlResponse
return parsers.JsonResponseParser(
parsers.ParserFor.make(TeacherUIUrlResponse)
).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(
self: RestrictionService[client.AuthenticatedClient],
*,
restriction_id: str,
) -> RestrictionWithEffectiveWindow:
"""Get the settings for a restriction plus the caller's entry window.
Gated by course visibility (`ensure_may_see`), so any enrolled student
can read it; the password it carries is viewer-relative. In addition to
the serialized restriction it returns `effective_entry_window`: the
requesting user's override-resolved entry window.
:param restriction_id: The id of the restriction.
:returns: The restriction object plus the caller's effective entry
window.
"""
url = "/api/v1/restrictions/{restrictionId}".format(
restrictionId=restriction_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.restriction_with_effective_window import (
RestrictionWithEffectiveWindow,
)
return parsers.JsonResponseParser(
parsers.ParserFor.make(RestrictionWithEffectiveWindow)
).try_parse(resp)
from ..models.any_error import AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)
[docs]
def patch(
self: RestrictionService[client.AuthenticatedClient],
json_body: PatchRestrictionData,
*,
restriction_id: str,
) -> Restriction:
"""Update restriction settings.
The request body mirrors the restriction JSON shape. All fields are
optional; omitted fields are left unchanged.
`proctoring` attaches or detaches an exam environment provider.
Attaching provisions the exam at the provider (and requires a
configured entry window); detaching archives it at the provider while
retaining its state, so re-attaching revives the same provider-side
exam. Provider state is committed atomically with the rest of the
patch: a provider failure fails the whole request.
:param json_body: The body of the request. See
:class:`.PatchRestrictionData` for information about the possible
fields. You can provide this data as a
:class:`.PatchRestrictionData` or as a dictionary.
:param restriction_id: The id of the restriction.
:returns: The updated restriction.
"""
url = "/api/v1/restrictions/{restrictionId}".format(
restrictionId=restriction_id
)
params = None
with self.__client as client:
resp = client.http.patch(
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.restriction import Restriction
return parsers.JsonResponseParser(
parsers.ParserFor.make(Restriction)
).try_parse(resp)
from ..models.any_error import AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)
[docs]
def launch_exam_environment(
self: RestrictionService[client.AuthenticatedClient],
json_body: LaunchExamEnvironmentRestrictionData,
*,
restriction_id: str,
) -> KioskLaunch:
"""Launch the exam environment for a non-direct entry ticket.
Provisions this launch's workspace at the provider — with the ticket
embedded in its entry URL, so the kiosk can activate it from inside —
and returns the redirect that sends the student into the kiosk. Nothing
is stored: relaunching (e.g. after a kiosk crash) provisions afresh
with a newly minted ticket.
The ticket must be an unactivated ticket for this restriction, created
by the calling user. It is not consumed here; it is activated later
through `PUT /restrictions/<id>/entries/` from inside the kiosk.
:param json_body: The body of the request. See
:class:`.LaunchExamEnvironmentRestrictionData` for information
about the possible fields. You can provide this data as a
:class:`.LaunchExamEnvironmentRestrictionData` or as a dictionary.
:param restriction_id: The id of the restriction to launch into.
:returns: The kiosk launch payload with the redirect to follow.
"""
url = "/api/v1/restrictions/{restrictionId}/exam_environment/launch".format(
restrictionId=restriction_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.kiosk_launch import KioskLaunch
return parsers.JsonResponseParser(
parsers.ParserFor.make(KioskLaunch)
).try_parse(resp)
from ..models.any_error import AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)
[docs]
def patch_entry_override(
self: RestrictionService[client.AuthenticatedClient],
json_body: PatchEntryOverrideRestrictionData,
*,
restriction_id: str,
user_id: int,
) -> EntryOverviewEntry:
"""Set or remove a per-student entry override for a restriction.
The override row is removed entirely (the student falls back to the
global `session_lockdown` settings) when the resolved limit and window
are both `null`; any non-null field creates or updates the override.
Both fields follow the standard PATCH convention: omit one to leave the
stored value unchanged, send `null` to clear it, or send a value to set
it. `override_entry_window` carries both bounds together, so they are
always set or cleared as a pair, and a set window must satisfy the same
ordering and maximum-duration rules as the global window. A submitted
window is ignored when the entry-window feature is disabled for the
tenant, mirroring `patch_restriction`.
:param json_body: The body of the request. See
:class:`.PatchEntryOverrideRestrictionData` for information about
the possible fields. You can provide this data as a
:class:`.PatchEntryOverrideRestrictionData` or as a dictionary.
:param restriction_id: The ID of the restriction.
:param user_id: The ID of the student.
:returns: The updated entry overview for this student.
"""
url = "/api/v1/restrictions/{restrictionId}/entries/{userId}".format(
restrictionId=restriction_id, userId=user_id
)
params = None
with self.__client as client:
resp = client.http.patch(
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.entry_overview_entry import EntryOverviewEntry
return parsers.JsonResponseParser(
parsers.ParserFor.make(EntryOverviewEntry)
).try_parse(resp)
from ..models.any_error import AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)
[docs]
def heartbeat(
self: RestrictionService[client.AuthenticatedClient],
) -> t.Union[
RefreshedHeartbeatResponse,
VerificationFailedHeartbeatResponse,
VerificationUnavailableHeartbeatResponse,
]:
"""Send a heartbeat to keep an isolated exam session alive.
When heartbeat enforcement is active, the session's `expires_at` is set
to a short window (5 minutes). This endpoint extends it forward, capped
by `max_expires_at`.
For a restriction running in an exam environment the heartbeat must
also carry a valid kiosk signature. On a rejected or uncheckable
signature the session is not invalidated, but the refresh is skipped —
a student who left the kiosk has their session lapse within one
heartbeat cycle — and the response tag tells the client to warn the
student.
If the session is already expired, the auth flow rejects the request
before it reaches this endpoint (401).
:returns: The tagged heartbeat outcome with the session's `expires_at`.
"""
url = "/api/v1/restrictions/heartbeat"
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, 200):
from ..models.refreshed_heartbeat_response import (
RefreshedHeartbeatResponse,
)
from ..models.verification_failed_heartbeat_response import (
VerificationFailedHeartbeatResponse,
)
from ..models.verification_unavailable_heartbeat_response import (
VerificationUnavailableHeartbeatResponse,
)
return parsers.JsonResponseParser(
parsers.make_union(
parsers.ParserFor.make(RefreshedHeartbeatResponse),
parsers.ParserFor.make(
VerificationFailedHeartbeatResponse
),
parsers.ParserFor.make(
VerificationUnavailableHeartbeatResponse
),
)
).try_parse(resp)
from ..models.any_error import AnyErrorParser
raise utils.get_error(
resp, (((400, 409, 401, 403, 404, 429, 500), AnyErrorParser),)
)