only useful methods

Author: CodeBooster97

audiostack/content/file.py

@@ -14,30 +14,34 @@ class File:
     FAMILY = "v3"
     interface = RequestInterface(family=FAMILY)
 
-    class Item(APIResponseItem):
+    class Item:
         def __init__(self, response: dict) -> None:
-            super().__init__(response)
-            self.file_id = response["fileId"]
-            self.file_name = response["fileName"]
-            self.folder_id = response["folderId"]
-            self.content_type = response["mimeType"]
+            self.file_id: str = response["fileId"]
+            self.file_name: str = response["fileName"]
+            self.url: str = response.get("url", "")
+            self.created_by: str = response.get("createdBy", "")
+            self.last_modified: str = response.get("lastModified", "")
+            self.file_type: dict = response.get("fileType", {})
+            self.file_category: Optional[str] = response.get("fileCategory", None)
+            self.size: str = str(response.get("size", ""))
+            self.created_at: str = response.get("createdAt", "")
+            self.status: str = response.get("status", "")
+            self.duration: Optional[str] = response.get("duration", None)
 
         def download(self, fileName: str, path: str = "./") -> None:
-            if not fileName:
-                raise Exception("Please supply a valid file name")
             if not self.url:
-                raise Exception("No URL found for this file")
-            RequestInterface.download_url(self.url, destination=path, name=fileName)
+                raise Exception(
+                    "No URL found for this file. Please check the file has been processed."
+                )
+            RequestInterface.download_url(url=self.url, destination=path, name=fileName)
 
-    class List(APIResponseList):
-        def __init__(self, response: dict, list_type: str) -> None:
-            super().__init__(response, list_type)
-
-        def resolve_item(self, list_type: str, item: Any) -> "File.Item":
-            if list_type == "items" or list_type == "files":
-                return File.Item({"data": item})
-            else:
-                raise Exception()
+    @staticmethod
+    def get(file_id: str) -> Item:
+        r = File.interface.send_request(
+            rtype=RequestTypes.GET,
+            route=f"file/{file_id}",
+        )
+        return File.Item(response=r)
 
     @staticmethod
     def create(
@@ -61,140 +65,99 @@ def create(
         }
 
         if category:
-            payload["categoryId"] = File.get_category_id_by_name(category)
+            payload["categoryId"] = File.get_category_id_by_name(name=category)
 
         r = File.interface.send_request(
             rtype=RequestTypes.POST,
             route="file/create-upload-url",
             json=payload,
         )
-        File.interface.send_upload_request(local_path=local_path, upload_url=r["uploadUrl"], mime_type=r["mimeType"])
-        return File.Item(r)
-
-    @staticmethod
-    def modify(
-        file_id: str,
-        file_name: str,
-        category: str = "",
-    ) -> Item:
-        category_id = File.get_category_id_by_name(category)
-        payload = {
-            "fileName": file_name,
-            "categoryId": category_id,
-        }
-        r = File.interface.send_request(
-            rtype=RequestTypes.PATCH,
-            route=f"file/{file_id}",
-            json=payload,
+        File.interface.send_upload_request(
+            local_path=local_path, upload_url=r["uploadUrl"], mime_type=r["mimeType"]
         )
-        return File.Item(r)
 
-    @staticmethod
-    def get(file_id: str) -> Item:
-        r = File.interface.send_request(
-            rtype=RequestTypes.GET,
-            route=f"file/{file_id}",
-        )
-        return File.Item(r)
+        start = time.time()
+
+        file = File.get(file_id=r["fileId"])
+
+        while file.status != "uploaded" or time.time() - start >= TIMEOUT_THRESHOLD_S:
+            print("Response in progress please wait...")
+            file = File.get(file_id=r["fileId"])
+
+        if file.status != "uploaded":
+            raise Exception("File upload failed")
+
+        return file
 
     @staticmethod
-    def delete(file_id: str, folder_id: str) -> APIResponseItem:
-        # Needs more thought
-        r = File.interface.send_request(
+    def delete(file_id: str, folder_id: str = "") -> None:
+        if not folder_id:
+            folder_id = Folder.get_root_folder_id()
+
+        File.interface.send_request(
             rtype=RequestTypes.DELETE,
             route=f"file/{file_id}/{folder_id}",
         )
-        return APIResponseItem(r)
-
-    @staticmethod
-    def get_file_categories() -> APIResponseItem:
-        r = File.interface.send_request(
-            rtype=RequestTypes.GET,
-            route="file/metadata/file-categories",
-        )
-        return APIResponseItem(r)
-
-    @staticmethod
-    def get_category_id_by_name(name: str) -> Optional[UUID]:
-        category_id = None
-        categories = [
-            {"category_id": y["categoryId"], "name": y["name"]}
-            for x in File.get_file_categories().data["fileTypes"]
-            for y in x["fileCategories"]
-        ]
-        category_id = next(filter(lambda x: x["name"] == name, categories), None)
-        return category_id
 
 
 class Folder:
     FAMILY = "v3"
     interface = RequestInterface(family=FAMILY)
 
-    class Item(APIResponseItem):
+    class Item:
         def __init__(self, response: dict) -> None:
-            super().__init__(response)
-            self.folders = Folder.List(response["folders"], list_type="folders")
-            self.files = File.List(response["files"], list_type="files")
-            self.current_path_chain = response["currentPathChain"]
-
-    class List(APIResponseList):
-        def __init__(self, response: dict, list_type: str) -> None:
-            super().__init__(response, list_type)
-
-        def resolve_item(self, list_type: str, item: Any) -> dict:
-            if list_type == "folders":
-                return {"data": item}
-            else:
-                raise Exception()
+            self.folder_id: str = response["folderId"]
+            self.folder_name: str = response["folderName"]
+            self.parent_folder_id: str = response.get("parentFolderId", "")
+            self.created_by: str = response.get("createdBy", "")
+            self.last_modified: Optional[str] = response.get("lastModified", None)
+            self.created_at: str = response.get("createdAt", "")
+
+    class ListResponse:
+        def __init__(self, response: dict) -> None:
+            self.folders: list[Folder.Item] = [
+                Folder.Item(response=x) for x in response["folders"]
+            ]
+            self.files: list[File.Item] = [
+                File.Item(response=x) for x in response["files"]
+            ]
+            self.current_path_chain: dict = response["currentPathChain"]
 
     @staticmethod
-    def get_root() -> Item:
-        r = Folder.interface.send_request(
+    def get_root_folder_id() -> str:
+        root_folder_id = Folder.interface.send_request(
             rtype=RequestTypes.GET,
             route="folder",
-        )
-        return Folder.Item(r)
+        )["currentPathChain"][0]["folderId"]
+        return root_folder_id
 
     @staticmethod
-    def create(name: str, parent_folder_id: Optional[UUID] = None) -> APIResponseItem:
-        folder = {
+    def create(name: str, parent_folder_id: Optional[UUID] = None) -> "Folder.Item":
+        payload = {
             "folderName": name,
-            "parentFolderId": parent_folder_id,
         }
+
+        if parent_folder_id:
+            payload["parentFolderId"] = str(parent_folder_id)
+
         r = Folder.interface.send_request(
             rtype=RequestTypes.POST,
             route="folder",
-            json=folder,
+            json=payload,
         )
-        return APIResponseItem(r)
+        return Folder.Item(response=r)
 
     @staticmethod
-    def get(folder_id: UUID) -> APIResponseItem:
+    def get(folder_id: UUID) -> "Folder.Item":
         r = Folder.interface.send_request(
             rtype=RequestTypes.GET,
             route=f"folder/{folder_id}",
         )
-        return APIResponseItem(r)
+        return Folder.Item(response=r["currentPathChain"][0])
 
     @staticmethod
-    def modify(
-        folder_id: UUID,
-        name: str,
-    ) -> APIResponseItem:
-        folder = {
-            "folderName": name,
-        }
-        r = Folder.interface.send_request(
-            rtype=RequestTypes.PATCH,
-            route=f"folder/{folder_id}",
-            json=folder,
-        )
-        return APIResponseItem(r)
-
-    @staticmethod
-    def delete(folder_id: UUID) -> APIResponseItem:
-        r = File.interface.send_request(
+    def delete(folder_id: UUID) -> None:
+        File.interface.send_request(
             rtype=RequestTypes.DELETE,
             route=f"folder/{folder_id}",
         )
-        return APIResponseItem(r)

audiostack/helpers/request_interface.py

@@ -65,11 +65,14 @@ def resolve_response(self, r: Any) -> dict:
             raise Exception(exc)
 
         if r.status_code >= 400:
-            msg = (
-                r.json()["message"]
-                + ". Errors listed as follows: \n\t"
-                + "\t".join(r.json()["errors"])
-            )
+            if "message" in r.json():
+                msg = (
+                    r.json()["message"]
+                    + ". Errors listed as follows: \n\t"
+                    + "\t".join(r.json()["errors"])
+                )
+            else:
+                msg = r.json()
             raise Exception(msg)
 
         if "meta" in r.json():
@@ -78,16 +81,18 @@ def resolve_response(self, r: Any) -> dict:
 
         return {**r.json(), **{"statusCode": r.status_code}}
 
-    def send_upload_request(self, local_path: str, upload_url: str, mime_type: str) -> int:
+    def send_upload_request(
+        self, local_path: str, upload_url: str, mime_type: str
+    ) -> int:
         with open(local_path, "rb") as data:
-            r = requests.put(url=upload_url, data=data, headers={"Content-Type": mime_type})
-            print("Upload response:", r.text)
+            r = requests.put(
+                url=upload_url, data=data, headers={"Content-Type": mime_type}
+            )
             if r.status_code >= 400:
                 raise Exception("Failed to upload file")
 
             return r.status_code
 
-
     def send_request(
         self,
         rtype: str,

audiostack/speech/tts.py

@@ -13,7 +13,7 @@ class TTS:
 
     class Item(APIResponseItem):
         def __init__(self, response: dict) -> None:
-            super().__init__(response)
+            super().__init__(response=response)
             self.speechId = self.data["speechId"]
 
         def download(

audiostack/tests/content/test_file.py

@@ -1,9 +1,7 @@
 import os
 
-import pytest
-
 import audiostack
-from audiostack.content.file import File, Folder
+from audiostack.content.file import File
 
 audiostack.api_base = os.environ.get("AUDIO_STACK_DEV_URL", "https://v2.api.audio")
 audiostack.api_key = os.environ["AUDIO_STACK_DEV_KEY"]  # type: ignore
@@ -12,26 +10,17 @@
 
 
 def test_create() -> None:
-    r = File.create(local_path="example.mp3", file_name="test")
+    r = File.create(local_path="example.mp3", file_name="test.mp3")
     test_constants["fileId"] = r.file_id
     test_constants["fileName"] = r.file_name
     print(r)
 
 
 def test_get() -> None:
-    r = File.get(test_constants["fileId"])
-    print(r)
-
-
-def test_modify() -> None:
-    r = File.modify(
-        file_id=test_constants["fileId"], file_name=test_constants["fileName"]
-    )
+    r = File.get(file_id=test_constants["fileId"])
     print(r)
 
 
 def test_delete() -> None:
-    r = File.delete(
-        test_constants["fileId"], Folder.get_root().current_path_chain["folderId"]
-    )
+    r = File.delete(file_id=test_constants["fileId"])
     print(r)

audiostack/tests/content/test_folder.py

@@ -1,4 +1,7 @@
+# TODO: User proper pytest fixtures
 import os
+import random
+import string
 
 import audiostack
 from audiostack.content.file import Folder
@@ -9,9 +12,20 @@
 test_constants: dict = {}  #
 
 
+def create_test_folder_name() -> str:
+    return "__PYTHON_TEST_" + "".join(
+        random.choice(string.ascii_letters) for _ in range(10)
+    )
+
+
 def test_create() -> None:
-    r = Folder.create(name="__PYTHON_TEST")
-    test_constants["folder_id"] = r.data["folderId"]
+    r = Folder.create(name=create_test_folder_name())
+    test_constants["folder_id"] = r.folder_id
+    print(r)
+
+
+def test_get() -> None:
+    r = Folder.get(folder_id=test_constants["folder_id"])
     print(r)