99 lines
3.7 KiB
Python
99 lines
3.7 KiB
Python
from typing import Final
|
|
|
|
from pydantic import parse_obj_as
|
|
from pydantic.dataclasses import dataclass
|
|
|
|
from mdrsclient.api.base import BaseApi
|
|
from mdrsclient.api.utils import token_check
|
|
from mdrsclient.exceptions import UnexpectedException
|
|
from mdrsclient.models import File
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FileCreateResponse:
|
|
id: str
|
|
|
|
|
|
class FileApi(BaseApi):
|
|
ENTRYPOINT: Final[str] = "v2/file/"
|
|
|
|
def retrieve(self, id: str) -> File:
|
|
# 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)
|
|
return parse_obj_as(File, response.json())
|
|
|
|
def create(self, folder_id: str, path: str) -> str:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT
|
|
token_check(self.connection)
|
|
data = {"folder_id": folder_id}
|
|
try:
|
|
with open(path, mode="rb") as fp:
|
|
response = self.connection.post(url, data=data, files={"file": fp})
|
|
self._raise_response_error(response)
|
|
ret = parse_obj_as(FileCreateResponse, response.json())
|
|
except OSError:
|
|
raise UnexpectedException(f"Could not open `{path}` file.")
|
|
return ret.id
|
|
|
|
def update(self, file: File, path: str | None) -> bool:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + file.id + "/"
|
|
token_check(self.connection)
|
|
if path is not None:
|
|
# update file body
|
|
try:
|
|
with open(path, mode="rb") as fp:
|
|
response = self.connection.put(url, files={"file": fp})
|
|
except OSError:
|
|
raise UnexpectedException(f"Could not open `{path}` file.")
|
|
else:
|
|
# update metadata
|
|
data = {"name": file.name, "description": file.description}
|
|
response = self.connection.put(url, data=data)
|
|
self._raise_response_error(response)
|
|
return True
|
|
|
|
def destroy(self, file: File) -> bool:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + file.id + "/"
|
|
token_check(self.connection)
|
|
response = self.connection.delete(url)
|
|
self._raise_response_error(response)
|
|
return True
|
|
|
|
def move(self, file: File, folder_id: str) -> bool:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + file.id + "/move/"
|
|
data = {"folder": folder_id}
|
|
token_check(self.connection)
|
|
response = self.connection.post(url, data=data)
|
|
self._raise_response_error(response)
|
|
return True
|
|
|
|
def metadata(self, file: File) -> dict:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = self.ENTRYPOINT + file.id + "/metadata/"
|
|
token_check(self.connection)
|
|
response = self.connection.get(url)
|
|
self._raise_response_error(response)
|
|
return response.json()
|
|
|
|
def download(self, file: File, path: str) -> bool:
|
|
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
|
|
url = "v2/" + file.download_url
|
|
response = self.connection.get(url, stream=True)
|
|
self._raise_response_error(response)
|
|
try:
|
|
with open(path, "wb") as f:
|
|
for chunk in response.iter_content(chunk_size=4096):
|
|
if chunk:
|
|
f.write(chunk)
|
|
f.flush()
|
|
except PermissionError:
|
|
print(f"Cannot create file `{path}`: Permission denied.")
|
|
return True
|