orrisroot
3081d7fbb3
Fetching a locked folder before its password is sent returns 200 with only its listing fields, without metadata, sub_folders and path, so FoldersApi.retrieve failed with a pydantic ValidationError. The DOI route fetches the folder before it can see the lock, so a password-protected DOI folder could not be opened even with its password. - Add FolderLockedException, a subclass of UnauthorizedException, and raise it from retrieve on that short record and from the lock checks. - On a DOI folder, send the password and fetch it again when the first fetch finds it locked.
137 lines
5.7 KiB
Python
137 lines
5.7 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 FolderLockedException, 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)
|
|
data = response.json()
|
|
# A locked folder not yet opened with its password comes back with 200 but only
|
|
# its listing fields, without `metadata`, `sub_folders` and `path`.
|
|
if data.get("lock") and "sub_folders" not in data:
|
|
raise FolderLockedException(f"Folder `{data.get('name', id)}` is locked.")
|
|
ret = TypeAdapter(Folder).validate_python(data)
|
|
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()
|