orrisroot
e5c28835b8
A request can wait to be served for longer than the access token it was sent with lives, and comes back refused for a token that was valid when it left. Uploads that take minutes make that wait ordinary. - tell an expired token apart from any other refusal by the code the API returns, and retry only that one, once - prefer a token another process left behind over minting a second: it saves a round trip, and a rotating provider would retire one that is still in use. Assign it rather than read it, since the setter is what rewrites the session header the retry will carry - settle the token before the request so the one compared afterwards is the one that was actually sent - report an overloaded server when the second attempt is refused too, and let API errors out of the upload path with their own type
132 lines
5.3 KiB
Python
132 lines
5.3 KiB
Python
from typing import Any, Final
|
|
|
|
import requests
|
|
from pydantic import TypeAdapter
|
|
from pydantic.dataclasses import dataclass
|
|
|
|
from mdrsclient.api.base import BaseApi
|
|
from mdrsclient.api.utils import retry_on_expired_token, token_check
|
|
from mdrsclient.exceptions import UnauthorizedException
|
|
from mdrsclient.models import Folder, FolderSimple
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FoldersApiCreateResponse:
|
|
id: str
|
|
|
|
|
|
class FoldersApi(BaseApi):
|
|
ENTRYPOINT: Final[str] = "v3/folders/"
|
|
|
|
@retry_on_expired_token
|
|
def list(self, laboratory_id: int, path: str) -> list[FolderSimple]:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT
|
|
params: dict[str, str | int] = {"path": path, "laboratory_id": laboratory_id}
|
|
token_check(self.connection)
|
|
response = self.connection.get(url, params=params)
|
|
self._raise_response_error(response)
|
|
ret: list[FolderSimple] = []
|
|
for data in response.json():
|
|
ret.append(TypeAdapter(FolderSimple).validate_python(data))
|
|
return ret
|
|
|
|
@retry_on_expired_token
|
|
def retrieve(self, id: str) -> Folder:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + id + "/"
|
|
token_check(self.connection)
|
|
response = self.connection.get(url)
|
|
self._raise_response_error(response)
|
|
ret = TypeAdapter(Folder).validate_python(response.json())
|
|
return ret
|
|
|
|
@retry_on_expired_token
|
|
def create(self, name: str, parent_id: str) -> str:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT
|
|
data: dict[str, str | int] = {"name": name, "parent_id": parent_id, "description": "", "template_id": -1}
|
|
token_check(self.connection)
|
|
response = self.connection.post(url, data=data)
|
|
self._raise_response_error(response)
|
|
ret = TypeAdapter(FoldersApiCreateResponse).validate_python(response.json())
|
|
return ret.id
|
|
|
|
@retry_on_expired_token
|
|
def update(self, folder: FolderSimple) -> bool:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + folder.id + "/"
|
|
data: dict[str, str | int] = {
|
|
"name": folder.name,
|
|
"description": folder.description,
|
|
}
|
|
token_check(self.connection)
|
|
response = self.connection.put(url, data=data)
|
|
self._raise_response_error(response)
|
|
return True
|
|
|
|
@retry_on_expired_token
|
|
def destroy(self, id: str, recursive: bool) -> bool:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + id + "/"
|
|
params: dict[str, str | int] = {"recursive": str(recursive)}
|
|
token_check(self.connection)
|
|
response = self.connection.delete(url, params=params)
|
|
self._raise_response_error(response)
|
|
return True
|
|
|
|
@retry_on_expired_token
|
|
def auth(self, id: str, password: str) -> bool:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + id + "/auth/"
|
|
data: dict[str, str | int] = {"password": password}
|
|
token_check(self.connection)
|
|
response = self.connection.post(url, data=data)
|
|
if response.status_code == requests.codes.unauthorized:
|
|
raise UnauthorizedException("Password is incorrect.")
|
|
self._raise_response_error(response)
|
|
return True
|
|
|
|
@retry_on_expired_token
|
|
def acl(self, id: str, access_level: int, recursive: bool, password: str | None) -> bool:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + id + "/acl/"
|
|
data: dict[str, str | int] = {"access_level": access_level}
|
|
if password is not None:
|
|
data.update({"password": password})
|
|
if recursive is True:
|
|
data.update({"lower": 1})
|
|
token_check(self.connection)
|
|
response = self.connection.post(url, data=data)
|
|
self._raise_response_error(response)
|
|
return True
|
|
|
|
@retry_on_expired_token
|
|
def move(self, folder: FolderSimple, folder_id: str, name: str) -> bool:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + folder.id + "/move/"
|
|
data: dict[str, str | int] = {"parent": folder_id, "name": name}
|
|
token_check(self.connection)
|
|
response = self.connection.post(url, data=data)
|
|
self._raise_response_error(response)
|
|
return True
|
|
|
|
@retry_on_expired_token
|
|
def copy(self, folder: FolderSimple, folder_id: str, name: str) -> bool:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + folder.id + "/copy/"
|
|
data: dict[str, str | int] = {"parent": folder_id, "name": name}
|
|
token_check(self.connection)
|
|
response = self.connection.post(url, data=data)
|
|
self._raise_response_error(response)
|
|
return True
|
|
|
|
@retry_on_expired_token
|
|
def metadata(self, id: str) -> dict[str, Any]:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + id + "/metadata/"
|
|
token_check(self.connection)
|
|
response = self.connection.get(url)
|
|
self._raise_response_error(response)
|
|
return response.json()
|