from typing import Final import requests from pydantic import TypeAdapter from pydantic.dataclasses import dataclass from mdrsclient.api.base import BaseApi from mdrsclient.exceptions import UnauthorizedException from mdrsclient.models import Token, User @dataclass(frozen=True) class UserAuthResponse(Token): is_reviewer: bool | None = None laboratory: str | None = None lab_id: int | None = None class UserApi(BaseApi): ENTRYPOINT: Final[str] = "v2/" def auth(self, username: str, password: str) -> tuple[User, Token]: # print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name) url = self.ENTRYPOINT + "auth/" data: dict[str, str | int] = {"username": username, "password": password} response = self.connection.post(url, data=data) if response.status_code == requests.codes.unauthorized: raise UnauthorizedException("Invalid username or password.") self._raise_response_error(response) obj = TypeAdapter(UserAuthResponse).validate_python(response.json()) token = Token(access=obj.access, refresh=obj.refresh) laboratory_ids = [obj.lab_id] if obj.lab_id is not None else [] is_reviewer = obj.is_reviewer if obj.is_reviewer is not None else False user = User(id=token.user_id, username=username, laboratory_ids=laboratory_ids, is_reviewer=is_reviewer) return (user, token) def refresh(self, token: Token) -> Token: # print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name) url = self.ENTRYPOINT + "refresh/" data: dict[str, str | int] = {"refresh": token.refresh} response = self.connection.post(url, data=data) if response.status_code == requests.codes.unauthorized: raise UnauthorizedException("Token is invalid or expired.") self._raise_response_error(response) token = TypeAdapter(Token).validate_python(response.json()) return token