from dataclasses import field 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 Laboratory, Token, User @dataclass(frozen=True) class UserAuthResponse(Token): is_reviewer: bool | None = None laboratories: list[Laboratory] = field(default_factory=list) 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 = list(map(lambda x: x.id, obj.laboratories)) 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