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

import sys
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._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._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:
            try:
                with open(path, mode="rb") as fp:
                    response = self._put(url, files={"file": fp})
            except OSError:
                raise UnexpectedException(f"Could not open `{path}` file.")
        else:
            data = {"name": file.name, "description": file.description}
            response = self._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._delete(url)
        self._raise_response_error(response)
        return True

    def move(self, file: File, folder_id: str | None) -> 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._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._get(url)
        self._raise_response_error(response)
        return response.json()