Add support for Search URL

cloudinary/search.py

@@ -1,7 +1,10 @@
+import base64
 import json
 
+import cloudinary
 from cloudinary.api_client.call_api import call_json_api
-from cloudinary.utils import unique
+from cloudinary.utils import unique, unsigned_download_url_prefix, build_distribution_domain, base64url_encode, \
+    json_encode, compute_hex_hash, SIGNATURE_SHA256
 
 
 class Search(object):
@@ -15,7 +18,10 @@ class Search(object):
         'with_field': None,
     }
 
+    _ttl = 300  # Used for search URLs
+
     """Build and execute a search query."""
+
     def __init__(self):
         self.query = {}
 
@@ -51,6 +57,16 @@ def with_field(self, value):
         self._add("with_field", value)
         return self
 
+    def ttl(self, ttl):
+        """
+        Sets the time to live of the search URL.
+
+        :param ttl: The time to live in seconds.
+        :return: self
+        """
+        self._ttl = ttl
+        return self
+
     def to_json(self):
         return json.dumps(self.as_dict())
 
@@ -60,12 +76,6 @@ def execute(self, **options):
         uri = [self._endpoint, 'search']
         return call_json_api('post', uri, self.as_dict(), **options)
 
-    def _add(self, name, value):
-        if name not in self.query:
-            self.query[name] = []
-        self.query[name].append(value)
-        return self
-
     def as_dict(self):
         to_return = {}
 
@@ -77,6 +87,51 @@ def as_dict(self):
 
         return to_return
 
+    def to_url(self, ttl=None, next_cursor=None, **options):
+        """
+        Creates a signed Search URL that can be used on the client side.
+
+        :param ttl: The time to live in seconds.
+        :param next_cursor: Starting position.
+        :param options: Additional url delivery options.
+        :return: The resulting search URL.
+        """
+        api_secret = options.get("api_secret", cloudinary.config().api_secret or None)
+        if not api_secret:
+            raise ValueError("Must supply api_secret")
+
+        if ttl is None:
+            ttl = self._ttl
+
+        query = self.as_dict()
+
+        _next_cursor = query.pop("next_cursor", None)
+        if next_cursor is None:
+            next_cursor = _next_cursor
+
+        b64query = base64url_encode(json_encode(query, sort_keys=True))
+
+        prefix = build_distribution_domain(options)
+
+        signature = compute_hex_hash("{ttl}{b64query}{api_secret}".format(
+            ttl=ttl,
+            b64query=b64query,
+            api_secret=api_secret
+        ), algorithm=SIGNATURE_SHA256)
+
+        return "{prefix}/search/{signature}/{ttl}/{b64query}{next_cursor}".format(
+            prefix=prefix,
+            signature=signature,
+            ttl=ttl,
+            b64query=b64query,
+            next_cursor="/{}".format(next_cursor) if next_cursor else "")
+
     def endpoint(self, endpoint):
         self._endpoint = endpoint
         return self
+
+    def _add(self, name, value):
+        if name not in self.query:
+            self.query[name] = []
+        self.query[name].append(value)
+        return self

cloudinary/utils.py

@@ -281,15 +281,15 @@ def encode_context(context):
     return "|".join(("{}={}".format(k, normalize_context_value(v))) for k, v in iteritems(context))
 
 
-def json_encode(value):
+def json_encode(value, sort_keys=False):
     """
     Converts value to a json encoded string
 
     :param value: value to be encoded
 
     :return: JSON encoded string
     """
-    return json.dumps(value, default=__json_serializer, separators=(',', ':'))
+    return json.dumps(value, default=__json_serializer, separators=(',', ':'), sort_keys=sort_keys)
 
 
 def encode_date_to_usage_api_format(date_obj):
@@ -709,6 +709,25 @@ def unsigned_download_url_prefix(source, cloud_name, private_cdn, cdn_subdomain,
     return prefix
 
 
+def build_distribution_domain(options):
+    source = options.pop('source', '')
+    cloud_name = options.pop("cloud_name", cloudinary.config().cloud_name or None)
+    if cloud_name is None:
+        raise ValueError("Must supply cloud_name in tag or in configuration")
+    secure = options.pop("secure", cloudinary.config().secure)
+    private_cdn = options.pop("private_cdn", cloudinary.config().private_cdn)
+    cname = options.pop("cname", cloudinary.config().cname)
+    secure_distribution = options.pop("secure_distribution",
+                                      cloudinary.config().secure_distribution)
+    cdn_subdomain = options.pop("cdn_subdomain", cloudinary.config().cdn_subdomain)
+    secure_cdn_subdomain = options.pop("secure_cdn_subdomain",
+                                       cloudinary.config().secure_cdn_subdomain)
+
+    return unsigned_download_url_prefix(
+        source, cloud_name, private_cdn, cdn_subdomain, secure_cdn_subdomain,
+        cname, secure, secure_distribution)
+
+
 def merge(*dict_args):
     result = None
     for dictionary in dict_args:
@@ -737,19 +756,8 @@ def cloudinary_url(source, **options):
     version = options.pop("version", None)
 
     format = options.pop("format", None)
-    cdn_subdomain = options.pop("cdn_subdomain", cloudinary.config().cdn_subdomain)
-    secure_cdn_subdomain = options.pop("secure_cdn_subdomain",
-                                       cloudinary.config().secure_cdn_subdomain)
-    cname = options.pop("cname", cloudinary.config().cname)
     shorten = options.pop("shorten", cloudinary.config().shorten)
 
-    cloud_name = options.pop("cloud_name", cloudinary.config().cloud_name or None)
-    if cloud_name is None:
-        raise ValueError("Must supply cloud_name in tag or in configuration")
-    secure = options.pop("secure", cloudinary.config().secure)
-    private_cdn = options.pop("private_cdn", cloudinary.config().private_cdn)
-    secure_distribution = options.pop("secure_distribution",
-                                      cloudinary.config().secure_distribution)
     sign_url = options.pop("sign_url", cloudinary.config().sign_url)
     api_secret = options.pop("api_secret", cloudinary.config().api_secret)
     url_suffix = options.pop("url_suffix", None)
@@ -795,9 +803,9 @@ def cloudinary_url(source, **options):
             base64.urlsafe_b64encode(
                 hash_fn(to_bytes(to_sign + api_secret)).digest())[0:chars_length]) + "--"
 
-    prefix = unsigned_download_url_prefix(
-        source, cloud_name, private_cdn, cdn_subdomain, secure_cdn_subdomain,
-        cname, secure, secure_distribution)
+    options["source"] = source
+    prefix = build_distribution_domain(options)
+
     source = "/".join(__compact(
         [prefix, resource_type, type, signature, transformation, version, source]))
     if sign_url and auth_token:

test/test_search.py

@@ -11,7 +11,8 @@
 from cloudinary import uploader, SearchFolders, Search
 from test.helper_test import SUFFIX, TEST_IMAGE, TEST_TAG, UNIQUE_TAG, TEST_FOLDER, UNIQUE_TEST_FOLDER, \
     retry_assertion, cleanup_test_resources_by_tag
-from test.test_api import MOCK_RESPONSE
+from test.test_api import MOCK_RESPONSE, NEXT_CURSOR
+from test.test_config import CLOUD_NAME, API_KEY, API_SECRET
 
 TEST_TAG = 'search_{}'.format(TEST_TAG)
 UNIQUE_TAG = 'search_{}'.format(UNIQUE_TAG)
@@ -60,6 +61,9 @@ def setUp(self):
     def tearDownClass(cls):
         cleanup_test_resources_by_tag([(UNIQUE_TAG,)])
 
+    def tearDown(self):
+        cloudinary.reset_config()
+
     @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
     def test_should_create_empty_json(self):
 
@@ -226,6 +230,84 @@ def test_should_not_duplicate_values(self, mocker):
             'with_field': ['context', 'tags'],
         })
 
+    def test_should_build_search_url(self):
+        cloudinary.config(cloud_name=CLOUD_NAME, api_key=API_KEY, api_secret=API_SECRET, secure=True)
+
+        search = Search() \
+            .expression("resource_type:image AND tags=kitten AND uploaded_at>1d AND bytes>1m") \
+            .sort_by("public_id", "desc") \
+            .max_results("30")
+
+        b64query = "eyJleHByZXNzaW9uIjoicmVzb3VyY2VfdHlwZTppbWFnZSBBTkQgdGFncz1raXR0ZW4gQU5EIHVwbG9hZGVkX2F0PjFkIEFO" \
+                   "RCBieXRlcz4xbSIsIm1heF9yZXN1bHRzIjoiMzAiLCJzb3J0X2J5IjpbeyJwdWJsaWNfaWQiOiJkZXNjIn1dfQ=="
+
+        ttl300_sig = "eadda21336fcce66ce195cce1b57cddd66a8e475ba151c39174133264278d5a5"
+        ttl1000_sig = "63091d184c88299dd2c7b0235560a6c119c5beb22eefd94401104060b436b334"
+
+        # default usage
+        self.assertEqual("https://res.cloudinary.com/{cloud}/search/{sig}/{ttl}/{query}".format(
+            cloud=CLOUD_NAME,
+            sig=ttl300_sig,
+            ttl=300,
+            query=b64query
+        ),
+            search.to_url()
+        )
+
+        # same signature with next cursor
+        self.assertEqual("https://res.cloudinary.com/{cloud}/search/{sig}/{ttl}/{query}/{cursor}".format(
+            cloud=CLOUD_NAME,
+            sig=ttl300_sig,
+            ttl=300,
+            query=b64query,
+            cursor=NEXT_CURSOR
+        ),
+            search.to_url(next_cursor=NEXT_CURSOR)
+        )
+
+        # with custom ttl and next cursor
+        self.assertEqual("https://res.cloudinary.com/{cloud}/search/{sig}/{ttl}/{query}/{cursor}".format(
+            cloud=CLOUD_NAME,
+            sig=ttl1000_sig,
+            ttl=1000,
+            query=b64query,
+            cursor=NEXT_CURSOR
+        ),
+            search.to_url(ttl=1000, next_cursor=NEXT_CURSOR)
+        )
+
+        # ttl and cursor are set from the class
+        self.assertEqual("https://res.cloudinary.com/{cloud}/search/{sig}/{ttl}/{query}/{cursor}".format(
+            cloud=CLOUD_NAME,
+            sig=ttl1000_sig,
+            ttl=1000,
+            query=b64query,
+            cursor=NEXT_CURSOR
+        ),
+            search.ttl(1000).next_cursor(NEXT_CURSOR).to_url()
+        )
+
+        # private cdn
+        self.assertEqual("https://{cloud}-res.cloudinary.com/search/{sig}/{ttl}/{query}".format(
+            cloud=CLOUD_NAME,
+            sig=ttl300_sig,
+            ttl=300,
+            query=b64query
+        ),
+            search.to_url(ttl=300, next_cursor="", private_cdn=True)
+        )
+
+        # private cdn from config
+        cloudinary.config(private_cdn=True)
+        self.assertEqual("https://{cloud}-res.cloudinary.com/search/{sig}/{ttl}/{query}".format(
+            cloud=CLOUD_NAME,
+            sig=ttl300_sig,
+            ttl=300,
+            query=b64query
+        ),
+            search.to_url(ttl=300, next_cursor="")
+        )
+
     @patch('urllib3.request.RequestMethods.request')
     def test_should_search_folders_endpoint(self, mocker):
         mocker.return_value = MOCK_RESPONSE