Api implementation

imagekitio/file.py

@@ -1,3 +1,4 @@
+import ast
 from json import dumps
 from typing import Any, Dict
 
@@ -162,10 +163,10 @@ def get_file_versions(self, file_id: str = None) -> ListFileResult:
         elif resp.status_code == 404:
             response_json = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = (
-                response_json["message"] if type(response_json) == dict else ""
-            )
-            response_help = response_json["help"] if type(response_json) == dict else ""
+            if type(response_json) == str:
+                response_json = ast.literal_eval(response_json)
+            error_message = response_json['message'] if type(response_json) == dict else ""
+            response_help = response_json['help'] if type(response_json) == dict else ""
             raise NotFoundException(error_message, response_help, response_meta_data)
         else:
             general_api_throw_exception(resp)
@@ -191,10 +192,10 @@ def get_file_version_details(
         elif resp.status_code == 404:
             response_json = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = (
-                response_json["message"] if type(response_json) == dict else ""
-            )
-            response_help = response_json["help"] if type(response_json) == dict else ""
+            if type(response_json) == str:
+                response_json = ast.literal_eval(response_json)
+            error_message = response_json['message'] if type(response_json) == dict else ""
+            response_help = response_json['help'] if type(response_json) == dict else ""
             raise NotFoundException(error_message, response_help, response_meta_data)
         else:
             general_api_throw_exception(resp)
@@ -288,8 +289,10 @@ def delete_file_version(self, file_id, version_id) -> ResponseMetadataResult:
         elif resp.status_code == 400 or resp.status_code == 404:
             response = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = response["message"] if type(response) == dict else ""
-            response_help = response["help"] if type(response) == dict else ""
+            if type(response) == str:
+                response = ast.literal_eval(response)
+            error_message = response['message'] if type(response) == dict else ""
+            response_help = response['help'] if type(response) == dict else ""
             if resp.status_code == 400:
                 raise BadRequestException(
                     error_message, response_help, response_meta_data
@@ -337,8 +340,10 @@ def copy_file(
         elif resp.status_code == 404:
             response = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = response["message"] if type(response) == dict else ""
-            response_help = response["help"] if type(response) == dict else ""
+            if type(response) == str:
+                response = ast.literal_eval(response)
+            error_message = response['message'] if type(response) == dict else ""
+            response_help = response['help'] if type(response) == dict else ""
             raise NotFoundException(error_message, response_help, response_meta_data)
         else:
             general_api_throw_exception(resp)
@@ -360,8 +365,10 @@ def move_file(
         elif resp.status_code == 404:
             response = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = response["message"] if type(response) == dict else ""
-            response_help = response["help"] if type(response) == dict else ""
+            if type(response) == str:
+                response = ast.literal_eval(response)
+            error_message = response['message'] if type(response) == dict else ""
+            response_help = response['help'] if type(response) == dict else ""
             raise NotFoundException(error_message, response_help, response_meta_data)
         else:
             general_api_throw_exception(resp)
@@ -383,8 +390,10 @@ def rename_file(self, options: RenameFileRequestOptions = None) -> RenameFileRes
         elif resp.status_code == 409:
             response = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = response["message"] if type(response) == dict else ""
-            response_help = response["help"] if type(response) == dict else ""
+            if type(response) == str:
+                response = ast.literal_eval(response)
+            error_message = response['message'] if type(response) == dict else ""
+            response_help = response['help'] if type(response) == dict else ""
             raise ConflictException(error_message, response_help, response_meta_data)
         else:
             general_api_throw_exception(resp)
@@ -408,8 +417,10 @@ def restore_file_version(
         elif resp.status_code == 404:
             response = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = response["message"] if type(response) == dict else ""
-            response_help = response["help"] if type(response) == dict else ""
+            if type(response) == str:
+                response = ast.literal_eval(response)
+            error_message = response['message'] if type(response) == dict else ""
+            response_help = response['help'] if type(response) == dict else ""
             raise NotFoundException(error_message, response_help, response_meta_data)
         else:
             general_api_throw_exception(resp)
@@ -430,8 +441,10 @@ def create_folder(
         else:
             response = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = response["message"] if type(response) == dict else ""
-            response_help = response["help"] if type(response) == dict else ""
+            if type(response) == str:
+                response = ast.literal_eval(response)
+            error_message = response['message'] if type(response) == dict else ""
+            response_help = response['help'] if type(response) == dict else ""
             raise UnknownException(error_message, response_help, response_meta_data)
 
     def delete_folder(
@@ -450,10 +463,10 @@ def delete_folder(
         elif resp.status_code == 404:
             response_json = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = (
-                response_json["message"] if type(response_json) == dict else ""
-            )
-            response_help = response_json["help"] if type(response_json) == dict else ""
+            if type(response_json) == str:
+                response_json = ast.literal_eval(response_json)
+            error_message = response_json['message'] if type(response_json) == dict else ""
+            response_help = response_json['help'] if type(response_json) == dict else ""
             raise NotFoundException(error_message, response_help, response_meta_data)
         else:
             general_api_throw_exception(resp)
@@ -473,10 +486,10 @@ def copy_folder(self, options: CopyFolderRequestOptions = None) -> FolderResult:
         elif resp.status_code == 404:
             response_json = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = (
-                response_json["message"] if type(response_json) == dict else ""
-            )
-            response_help = response_json["help"] if type(response_json) == dict else ""
+            if type(response_json) == str:
+                response_json = ast.literal_eval(response_json)
+            error_message = response_json['message'] if type(response_json) == dict else ""
+            response_help = response_json['help'] if type(response_json) == dict else ""
             raise NotFoundException(error_message, response_help, response_meta_data)
         else:
             general_api_throw_exception(resp)
@@ -497,10 +510,10 @@ def move_folder(self, options: MoveFolderRequestOptions = None) -> FolderResult:
         elif resp.status_code == 404:
             response_json = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = (
-                response_json["message"] if type(response_json) == dict else ""
-            )
-            response_help = response_json["help"] if type(response_json) == dict else ""
+            if type(response_json) == str:
+                response_json = ast.literal_eval(response_json)
+            error_message = response_json['message'] if type(response_json) == dict else ""
+            response_help = response_json['help'] if type(response_json) == dict else ""
             raise NotFoundException(error_message, response_help, response_meta_data)
         else:
             general_api_throw_exception(resp)
@@ -609,10 +622,10 @@ def create_custom_metadata_fields(
         else:
             response_json = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = (
-                response_json["message"] if type(response_json) == dict else ""
-            )
-            response_help = response_json["help"] if type(response_json) == dict else ""
+            if type(response_json) == str:
+                response_json = ast.literal_eval(response_json)
+            error_message = response_json['message'] if type(response_json) == dict else ""
+            response_help = response_json['help'] if type(response_json) == dict else ""
             if resp.status_code == 400:
                 raise BadRequestException(
                     error_message, response_help, response_meta_data
@@ -625,7 +638,7 @@ def get_custom_metadata_fields(
     ) -> ListCustomMetadataFieldsResult:
         """get custom metadata fields"""
         url = "{}/v1/customMetadataFields".format(URL.API_BASE_URL)
-        param = {"includeDeleted": include_deleted}
+        param = {"includeDeleted": str(include_deleted).lower()}
         resp = self.request.request(
             method="GET", url=url, headers=self.request.create_headers(), params=param
         )
@@ -637,8 +650,10 @@ def get_custom_metadata_fields(
         else:
             response = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = response["message"] if type(response) == dict else ""
-            response_help = response["help"] if type(response) == dict else ""
+            if type(response) == str:
+                response = ast.literal_eval(response)
+            error_message = response['message'] if type(response) == dict else ""
+            response_help = response['help'] if type(response) == dict else ""
             raise UnknownException(error_message, response_help, response_meta_data)
 
 
@@ -668,10 +683,10 @@ def update_custom_metadata_fields(
         else:
             response_json = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = (
-                response_json["message"] if type(response_json) == dict else ""
-            )
-            response_help = response_json["help"] if type(response_json) == dict else ""
+            if type(response_json) == str:
+                response_json = ast.literal_eval(response_json)
+            error_message = response_json['message'] if type(response_json) == dict else ""
+            response_help = response_json['help'] if type(response_json) == dict else ""
             if resp.status_code == 400:
                 raise BadRequestException(
                     error_message, response_help, response_meta_data
@@ -695,10 +710,10 @@ def delete_custom_metadata_field(self, field_id: str) -> ResponseMetadataResult:
         else:
             response_json = get_response_json(resp)
             response_meta_data = populate_response_metadata(resp)
-            error_message = (
-                response_json["message"] if type(response_json) == dict else ""
-            )
-            response_help = response_json["help"] if type(response_json) == dict else ""
+            if type(response_json) == str:
+                response_json = ast.literal_eval(response_json)
+            error_message = response_json['message'] if type(response_json) == dict else ""
+            response_help = response_json['help'] if type(response_json) == dict else ""
             if resp.status_code == 404:
                 raise NotFoundException(
                     error_message, response_help, response_meta_data

imagekitio/models/results/FileResult.py

@@ -12,7 +12,7 @@ def __init__(
         updated_at,
         file_id,
         tags,
-        ai_tags: List[dict] = [],
+        ai_tags: List[AITags] = [],
         version_info: dict = {},
         embedded_metadata: dict = {},
         custom_coordinates: str = "",
@@ -38,7 +38,9 @@ def __init__(
         self.ai_tags: List[AITags] = []
         if ai_tags is not None:
             for i in ai_tags:
-                self.ai_tags.append(AITags(i["name"], i["confidence"], i["source"]))
+                self.ai_tags.append(AITags(i["name"] if "name" in i else None,
+                                           i["confidence"] if "confidence" in i else None,
+                                           i["source"] if 'source' in i else None))
         self.version_info = VersionInfo(version_info["id"], version_info["name"])
         self.embedded_metadata = embedded_metadata
         self.custom_coordinates = custom_coordinates

imagekitio/models/results/FileResultWithResponseMetadata.py

@@ -11,10 +11,10 @@ def __init__(
         self,
         type,
         name,
-        created_at,
-        updated_at,
-        file_id,
-        tags,
+        created_at=None,
+        updated_at=None,
+        file_id=None,
+        tags=None,
         ai_tags: List[AITags] = AITags(None, None, None),
         version_info: VersionInfo = VersionInfo(None, None),
         embedded_metadata=None,

imagekitio/utils/utils.py

@@ -1,3 +1,4 @@
+import ast
 from json import loads, dumps
 from requests.models import Response
 
@@ -36,8 +37,10 @@ def populate_response_metadata(response: Response):
 def general_api_throw_exception(response: Response):
     resp = get_response_json(response)
     response_meta_data = populate_response_metadata(response)
-    error_message = resp["message"] if type(resp) == dict else ""
-    response_help = resp["help"] if type(resp) == dict and "help" in resp else ""
+    if type(resp) == str:
+        resp = ast.literal_eval(resp)
+    error_message = resp['message'] if type(resp) == dict else ""
+    response_help = resp['help'] if type(resp) == dict and 'help' in resp else ""
     if response.status_code == 400:
         raise BadRequestException(error_message, response_help, response_meta_data)
     elif response.status_code == 401:
@@ -60,8 +63,10 @@ def general_api_throw_exception(response: Response):
 def throw_other_exception(response: Response):
     resp = get_response_json(response)
     response_meta_data = populate_response_metadata(response)
-    error_message = resp["message"] if type(resp) == dict else ""
-    response_help = resp["help"] if type(resp) == dict else ""
+    if type(resp) == str:
+        resp = ast.literal_eval(resp)
+    error_message = resp['message'] if type(resp) == dict else ""
+    response_help = resp['help'] if type(resp) == dict else ""
     if response.status_code == 207:
         raise PartialSuccessException(error_message, response_help, response_meta_data)
     elif response.status_code == 404:

requirements/test.txt

@@ -2,5 +2,5 @@ requests>=2.22.0
 black==19.10b0
 coverage==4.5.4
 tox==3.14.2
-responses==0.21.0
+responses==0.17.0
 requests_toolbelt==0.9.1

tests/helpers.py

@@ -1,4 +1,5 @@
 import base64
+import re
 import unittest
 from unittest.mock import patch
 
@@ -38,3 +39,7 @@ def get_auth_headers_for_test():
         "utf-8"
     )
     return {"Authorization": "Basic {}".format(encoded_private_key)}
+
+
+def make_string_to_single_line(multiline_string):
+    return re.sub(r'\s(?=\s)', '', re.sub(r'\s', ' ', multiline_string)).replace('{ ', '{').replace(' }', '}')