Add support for analyze API

Author: const-cloudinary

cloudinary/api.py

@@ -14,7 +14,8 @@
 from cloudinary.api_client.call_api import (
     call_api,
     call_metadata_api,
-    call_json_api
+    call_json_api,
+    _call_v2_api
 )
 from cloudinary.exceptions import (
     BadRequest,
@@ -840,3 +841,18 @@ def reorder_metadata_fields(order_by, direction=None, **options):
     uri = ['order']
     params = {'order_by': order_by, 'direction': direction}
     return call_metadata_api('put', uri, params, **options)
+
+
+def analyze(input_type, analysis_type, uri=None, **options):
+    """Analyzes an asset with the requested analysis type.
+
+    :param input_type: The type of input for the asset to analyze ('uri').
+    :param analysis_type: The type of analysis to run ('google_tagging', 'captioning', 'fashion').
+    :param uri: The URI of the asset to analyze.
+    :param options: Additional options.
+
+    :rtype: Response
+    """
+    api_uri = ['analysis', 'analyze']
+    params = {'input_type': input_type, 'analysis_type': analysis_type, 'uri': uri}
+    return _call_v2_api('post', api_uri, params, **options)

cloudinary/api_client/call_api.py

@@ -26,6 +26,10 @@ def call_json_api(method, uri, json_body, **options):
     return _call_api(method, uri, body=data, headers={'Content-Type': 'application/json'}, **options)
 
 
+def _call_v2_api(method, uri, json_body, **options):
+    return call_json_api(method, uri, json_body=json_body, api_version='v2', **options)
+
+
 def call_api(method, uri, params, **options):
     return _call_api(method, uri, params=params, **options)
 
@@ -42,10 +46,11 @@ def _call_api(method, uri, params=None, body=None, headers=None, extra_headers=N
     oauth_token = options.pop("oauth_token", cloudinary.config().oauth_token)
 
     _validate_authorization(api_key, api_secret, oauth_token)
-
-    api_url = "/".join([prefix, cloudinary.API_VERSION, cloud_name] + uri)
     auth = {"key": api_key, "secret": api_secret, "oauth_token": oauth_token}
 
+    api_version = options.pop("api_version",  cloudinary.API_VERSION)
+    api_url = "/".join([prefix, api_version, cloud_name] + uri)
+
     if body is not None:
         options["body"] = body
 

test/helper_test.py

@@ -48,6 +48,7 @@
     # urllib3 2.x support
     # noinspection PyProtectedMember
     import urllib3._request_methods
+
     URLLIB3_REQUEST = "urllib3._request_methods.RequestMethods.request"
 except ImportError:
     URLLIB3_REQUEST = "urllib3.request.RequestMethods.request"
@@ -140,10 +141,10 @@ def http_response_mock(body="", headers=None, status=200):
     return HTTPResponse(body, HTTPHeaderDict(headers), status=status)
 
 
-def api_response_mock():
-    return http_response_mock('{"foo":"bar"}', {"x-featureratelimit-limit": '0',
-                                                "x-featureratelimit-reset": 'Sat, 01 Apr 2017 22:00:00 GMT',
-                                                "x-featureratelimit-remaining": '0'})
+def api_response_mock(body='{"foo":"bar"}'):
+    return http_response_mock(body, {"x-featureratelimit-limit": '0',
+                                     "x-featureratelimit-reset": 'Sat, 01 Apr 2017 22:00:00 GMT',
+                                     "x-featureratelimit-remaining": '0'})
 
 
 def uploader_response_mock():

test/test_api.py

@@ -1218,6 +1218,26 @@ def test_structured_metadata_in_resources_by_moderation_api(self):
         for resource in result["resources"]:
             self.assertNotIn("metadata", resource)
 
+    @patch(URLLIB3_REQUEST)
+    def test_analyze(self, mocker):
+        mocker.return_value = MOCK_RESPONSE
+
+        options = {
+            "input_type": "uri",
+            "analysis_type": "captioning",
+            "uri": "https://res.cloudinary.com/demo/image/upload/dog",
+        }
+
+        api.analyze(**options)
+
+        uri = get_uri(mocker)
+        self.assertIn("/v2/", uri)
+        self.assertTrue(uri.endswith("/analysis/analyze"))
+
+        params = get_json_body(mocker)
+        for param in options.keys():
+            self.assertIn(param, params)
+
 
 if __name__ == '__main__':
     unittest.main()