Source code for codegrade._api.user

"""The endpoints for user objects.

SPDX-License-Identifier: AGPL-3.0-only OR BSD-3-Clause-Clear
"""

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 parsers, utils

if t.TYPE_CHECKING or os.getenv("CG_EAGERIMPORT", False):
    from .. import client
    from ..models.course_coupon_usage import CourseCouponUsage
    from ..models.extended_transaction import ExtendedTransaction
    from ..models.extended_user import ExtendedUser
    from ..models.login_user_data import LoginUserData
    from ..models.logout_response import LogoutResponse
    from ..models.logout_user_data import LogoutUserData
    from ..models.patch_user_data import PatchUserData
    from ..models.register_user_data import RegisterUserData
    from ..models.session_restriction_data import SessionRestrictionData
    from ..models.tenant_coupon_usage import TenantCouponUsage
    from ..models.user import User
    from ..models.user_login_response import UserLoginResponse


_ClientT = t.TypeVar("_ClientT", bound="client._BaseClient")


[docs]class UserService(t.Generic[_ClientT]): __slots__ = ("__client",) def __init__(self, client: _ClientT) -> None: self.__client = client
[docs] def patch( self: "UserService[client.AuthenticatedClient]", json_body: t.Union[dict, list, "PatchUserData"], *, user_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "ExtendedUser": """Update the attributes of a user. :param json_body: The body of the request. See :class:`.PatchUserData` for information about the possible fields. You can provide this data as a :class:`.PatchUserData` or as a dictionary. :param user_id: The id of the user you want to change. Currently this can only be your own user id. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The updated user. """ url = "/api/v1/users/{userId}".format(userId=user_id) params = extra_parameters or {} 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.extended_user import ExtendedUser return parsers.JsonResponseParser( parsers.ParserFor.make(ExtendedUser) ).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_coupon_usages( self: "UserService[client.AuthenticatedClient]", *, user_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "t.Sequence[t.Union[TenantCouponUsage, CourseCouponUsage]]": """Get all the coupons used for the specified user. :param user_id: The user to get the coupons used for. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: All coupons used for the given user. """ url = "/api/v1/users/{userId}/coupon_usages/".format(userId=user_id) params = extra_parameters or {} 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.course_coupon_usage import CourseCouponUsage from ..models.tenant_coupon_usage import TenantCouponUsage return parsers.JsonResponseParser( rqa.List( parsers.make_union( parsers.ParserFor.make(TenantCouponUsage), parsers.ParserFor.make(CourseCouponUsage), ) ) ).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_transactions( self: "UserService[client.AuthenticatedClient]", *, user_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "t.Sequence[ExtendedTransaction]": """Get all transactions for the specified user. :param user_id: The user to get the transactions for. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: All transactions for the given user. """ url = "/api/v1/users/{userId}/transactions/".format(userId=user_id) params = extra_parameters or {} 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_transaction import ExtendedTransaction return parsers.JsonResponseParser( rqa.List(parsers.ParserFor.make(ExtendedTransaction)) ).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: "UserService[client.AuthenticatedClient]", *, type: "t.Literal['default', 'extended', 'roles']" = "default", extended: "bool" = False, with_permissions: "bool" = False, extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "t.Union[ExtendedUser, User, t.Mapping[str, str]]": """Get the info of the currently logged in user. :param type: If this is `roles` a mapping between course_id and role name will be returned, if this is `extended` an `ExtendedUser` instead of a `User` will be returned. :param extended: If `true` this has the same effect as setting `type` to `extended`. :param with_permissions: Setting this to true will add the key `permissions` to the user. The value will be a mapping indicating which global permissions this user has. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: A response containing the JSON serialized user """ url = "/api/v1/login" params: t.Dict[str, t.Any] = { **(extra_parameters or {}), "type": utils.to_dict(type), "extended": utils.to_dict(extended), "with_permissions": utils.to_dict(with_permissions), } 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_user import ExtendedUser from ..models.user import User return parsers.JsonResponseParser( parsers.make_union( parsers.ParserFor.make(ExtendedUser), parsers.ParserFor.make(User), rqa.LookupMapping(rqa.SimpleValue.str), ) ).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 login( self, json_body: t.Union[dict, list, "LoginUserData"], *, with_permissions: "bool" = False, extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "UserLoginResponse": """Login using your username and password. `permissions` to the user. The value will be a mapping indicating which global permissions this user has. :param json_body: The body of the request. See :class:`.LoginUserData` for information about the possible fields. You can provide this data as a :class:`.LoginUserData` or as a dictionary. :param with_permissions: Return the global permissions of the newly logged in user. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: A response containing the JSON serialized user """ url = "/api/v1/login" params: t.Dict[str, t.Any] = { **(extra_parameters or {}), "with_permissions": utils.to_dict(with_permissions), } 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.user_login_response import UserLoginResponse return parsers.JsonResponseParser( parsers.ParserFor.make(UserLoginResponse) ).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 register( self, json_body: t.Union[dict, list, "RegisterUserData"], *, extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "UserLoginResponse": """Create a new user. :param json_body: The body of the request. See :class:`.RegisterUserData` for information about the possible fields. You can provide this data as a :class:`.RegisterUserData` or as a dictionary. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The registered user and an `access_token` that can be used to perform requests as this new user. """ url = "/api/v1/user" params = extra_parameters or {} 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.user_login_response import UserLoginResponse return parsers.JsonResponseParser( parsers.ParserFor.make(UserLoginResponse) ).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 restrict( self: "UserService[client.AuthenticatedClient]", json_body: t.Union[dict, list, "SessionRestrictionData"], *, extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "UserLoginResponse": """Revoke a given token. :param json_body: The body of the request. See :class:`.SessionRestrictionData` for information about the possible fields. You can provide this data as a :class:`.SessionRestrictionData` or as a dictionary. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: An new token that has the given restrictions added. """ url = "/api/v1/token/restrict" params = extra_parameters or {} 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.user_login_response import UserLoginResponse return parsers.JsonResponseParser( parsers.ParserFor.make(UserLoginResponse) ).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 logout( self, multipart_data: t.Union[dict, list, "LogoutUserData"], *, extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "LogoutResponse": """Revoke a given token. :param multipart_data: The data that should form the body of the request. See :class:`.LogoutUserData` for information about the possible fields. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: An empty 200 response. """ url = "/api/v1/token/revoke" params = extra_parameters or {} data, files = utils.to_multipart(utils.to_dict(multipart_data)) with self.__client as client: resp = client.http.post( url=url, files=files, data=data, params=params ) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.logout_response import LogoutResponse return parsers.JsonResponseParser( parsers.ParserFor.make(LogoutResponse) ).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 search( self: "UserService[client.AuthenticatedClient]", *, q: "str", exclude_course: Maybe["int"] = Nothing, extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "t.Sequence[User]": """Search for a user by name and username. :param q: The string to search for, all SQL wildcard are escaped and spaces are replaced by wildcards. :param exclude_course: Exclude all users that are in the given course from the search results. You need the permission `can_list_course_users` on this course to use this parameter. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The users that match the given query string. """ url = "/api/v1/users/" params: t.Dict[str, t.Any] = { **(extra_parameters or {}), "q": utils.to_dict(q), } maybe_from_nullable(t.cast(t.Any, exclude_course)).if_just( lambda val: params.__setitem__("exclude_course", val) ) 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.user import User return parsers.JsonResponseParser( rqa.List(parsers.ParserFor.make(User)) ).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), ), ), )