Add support for related assets APIs

Author: const-cloudinary

cloudinary/api.py

@@ -190,7 +190,7 @@ def _prepare_asset_details_params(**options):
     """
     return only(options, "exif", "faces", "colors", "image_metadata", "media_metadata", "cinemagraph_analysis",
                 "pages", "phash", "coordinates", "max_results", "quality_analysis", "derived_next_cursor",
-                "accessibility_analysis", "versions")
+                "accessibility_analysis", "versions", "related", "related_next_cursor")
 
 
 def update(public_id, **options):
@@ -295,6 +295,50 @@ def delete_derived_by_transformation(public_ids, transformations,
     return call_api("delete", uri, params, **options)
 
 
+def add_related_assets(public_id, assets_to_relate, resource_type="image", type="upload", **options):
+    """
+    Relates an asset to other assets by public IDs.
+
+    :param public_id: The public ID of the asset to update.
+    :type public_id: str
+    :param assets_to_relate: The array of up to 10 fully_qualified_public_ids given as resource_type/type/public_id.
+    :type assets_to_relate: list[str]
+    :param type: The upload type. Defaults to "upload".
+    :type type: str
+    :param resource_type: The type of the resource. Defaults to "image".
+    :type resource_type: str
+    :param options: Additional options.
+    :type options: dict, optional
+    :return: The result of the command.
+    :rtype: dict
+    """
+    uri = ["resources", "related_assets", resource_type, type, public_id]
+    params = {"assets_to_relate": utils.build_array(assets_to_relate)}
+    return call_json_api("post", uri, params, **options)
+
+
+def delete_related_assets(public_id, assets_to_unrelate, resource_type="image", type="upload", **options):
+    """
+    Unrelates an asset from other assets by public IDs.
+
+    :param public_id: The public ID of the asset to update.
+    :type public_id: str
+    :param assets_to_unrelate: The array of up to 10 fully_qualified_public_ids given as resource_type/type/public_id.
+    :type assets_to_unrelate: list[str]
+    :param type: The upload type.
+    :type type: str
+    :param resource_type: The type of the resource: defaults to "image".
+    :type resource_type: str
+    :param options: Additional options.
+    :type options: dict, optional
+    :return: The result of the command.
+    :rtype: dict
+    """
+    uri = ["resources", "related_assets", resource_type, type, public_id]
+    params = {"assets_to_unrelate": utils.build_array(assets_to_unrelate)}
+    return call_json_api("delete", uri, params, **options)
+
+
 def tags(**options):
     resource_type = options.pop("resource_type", "image")
     uri = ["tags", resource_type]

cloudinary/utils.py

@@ -1579,3 +1579,19 @@ def unique(collection, key=None):
         to_return[key(element)] = element
 
     return list(to_return.values())
+
+
+def fq_public_id(public_id, resource_type="image", type="upload"):
+    """
+    Returns the fully qualified public id of form resource_type/type/public_id.
+
+    :param public_id: The public ID of the asset.
+    :type public_id: str
+    :param resource_type: The type of the asset. Defaults to "image".
+    :type resource_type: str
+    :param type: The upload type. Defaults to "upload".
+    :type type: str
+
+    :return:
+    """
+    return "{resource_type}/{type}/{public_id}".format(resource_type=resource_type, type=type, public_id=public_id)

test/test_api.py

@@ -9,6 +9,7 @@
 
 import cloudinary
 from cloudinary import api, uploader, utils
+from cloudinary.utils import fq_public_id
 from test.helper_test import SUFFIX, TEST_IMAGE, get_uri, get_params, get_list_param, get_param, TEST_DOC, get_method, \
     UNIQUE_TAG, api_response_mock, ignore_exception, cleanup_test_resources_by_tag, cleanup_test_transformation, \
     cleanup_test_resources, UNIQUE_TEST_FOLDER, EVAL_STR, get_json_body
@@ -461,6 +462,33 @@ def test09_delete_resources_tuple(self, mocker):
         self.assertIn(API_TEST_ID, param)
         self.assertIn(API_TEST_ID2, param)
 
+    @patch('urllib3.request.RequestMethods.request')
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test_add_related_assets(self, mocker):
+        """ should allow adding related assets """
+        mocker.return_value = MOCK_RESPONSE
+        api.add_related_assets(API_TEST_ID, [fq_public_id(API_TEST_ID2), fq_public_id(API_TEST_ID3)])
+        args, kargs = mocker.call_args
+        self.assertEqual(args[0], 'POST')
+        self.assertTrue(get_uri(args).endswith('/resources/related_assets/image/upload/' + API_TEST_ID))
+        param = get_json_body(mocker)['assets_to_relate']
+        self.assertIn(fq_public_id(API_TEST_ID2), param)
+        self.assertIn(fq_public_id(API_TEST_ID3), param)
+
+    @patch('urllib3.request.RequestMethods.request')
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test_delete_related_assets(self, mocker):
+        """ should allow deleting related assets """
+        mocker.return_value = MOCK_RESPONSE
+        api.delete_related_assets(API_TEST_ID, [fq_public_id(API_TEST_ID2), fq_public_id(API_TEST_ID3)])
+        args, kargs = mocker.call_args
+        self.assertEqual(args[0], 'DELETE')
+        self.assertTrue(get_uri(args).endswith('/resources/related_assets/image/upload/' + API_TEST_ID))
+        param = get_json_body(mocker)['assets_to_unrelate']
+        self.assertIn(fq_public_id(API_TEST_ID2), param)
+        self.assertIn(fq_public_id(API_TEST_ID3), param)
+
+
     @patch('urllib3.request.RequestMethods.request')
     @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
     def test10_tags(self, mocker):
@@ -869,11 +897,11 @@ def test_delete_folder(self, mocker):
     def test_root_folders_allows_next_cursor_and_max_results_parameter(self, mocker):
         """ should allow next_cursor and max_results parameters """
         mocker.return_value = MOCK_RESPONSE
-        
+
         api.root_folders(next_cursor=NEXT_CURSOR, max_results=10)
-        
+
         args, kwargs = mocker.call_args
-        
+
         self.assertTrue("next_cursor" in get_params(args))
         self.assertTrue("max_results" in get_params(args))
 
@@ -882,11 +910,11 @@ def test_root_folders_allows_next_cursor_and_max_results_parameter(self, mocker)
     def test_subfolders_allows_next_cursor_and_max_results_parameter(self, mocker):
         """ should allow next_cursor and max_results parameters """
         mocker.return_value = MOCK_RESPONSE
-        
+
         api.subfolders(API_TEST_ID, next_cursor=NEXT_CURSOR, max_results=10)
-        
+
         args, kwargs = mocker.call_args
-        
+
         self.assertTrue("next_cursor" in get_params(args))
         self.assertTrue("max_results" in get_params(args))
 
@@ -996,26 +1024,22 @@ def test_update_access_control(self, mocker):
         self.assertEqual(exp_acl, params["access_control"])
 
     @patch('urllib3.request.RequestMethods.request')
-    def test_cinemagraph_analysis_resource(self, mocker):
-        """ should allow the user to pass cinemagraph_analysis in the resource function """
+    def test_various_resource_parameters(self, mocker):
+        """ should allow the user to pass various parameters to the resource function """
         mocker.return_value = MOCK_RESPONSE
 
-        api.resource(API_TEST_ID, cinemagraph_analysis=True)
-
-        params = get_params(mocker.call_args[0])
-
-        self.assertIn("cinemagraph_analysis", params)
-
-    @patch('urllib3.request.RequestMethods.request')
-    def test_accessibility_analysis_resource(self, mocker):
-        """ should allow the user to pass accessibility_analysis in the resource function """
-        mocker.return_value = MOCK_RESPONSE
+        options = {
+            "cinemagraph_analysis": True,
+            "accessibility_analysis": True,
+            "related": True,
+            "related_next_cursor": NEXT_CURSOR,
+        }
 
-        api.resource(API_TEST_ID, accessibility_analysis=True)
+        api.resource(TEST_IMAGE, **options)
 
         params = get_params(mocker.call_args[0])
-
-        self.assertIn("accessibility_analysis", params)
+        for param in options.keys():
+            self.assertIn(param, params)
 
     @patch('urllib3.request.RequestMethods.request')
     def test_api_url_escapes_special_characters(self, mocker):