mdrs-client-python/mdrsclient/api/user.py

47 lines
1.9 KiB
Python

from typing import Final
import requests
from pydantic import parse_obj_as
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 = {"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 = parse_obj_as(UserAuthResponse, 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 = {"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 = parse_obj_as(Token, response.json())
return token