Merge pull request #11 from imagekit-developer/url_refactor

Refactor url generation logic

Author: imagekitio

imagekitio/constants/defaults.py

@@ -8,5 +8,12 @@ class Default(enum.Enum):
         DEFAULT_TRANSFORMATION_POSITION,
         QUERY_TRANSFORMATION_POSITION,
     ]
-    DEFAULT_TIMESTAMP = "9999999999"
+    DEFAULT_TIMESTAMP = 9999999999
+    SDK_VERSION_PARAMETER = "ik-sdk-version"
     SDK_VERSION = "python-2.2.4"
+    TRANSFORMATION_PARAMETER = "tr"
+    CHAIN_TRANSFORM_DELIMITER = ":"
+    TRANSFORM_DELIMITER = ","
+    TRANSFORM_KEY_VALUE_DELIMITER = "-"
+    SIGNATURE_PARAMETER = "ik-s"
+    TIMESTAMP_PARAMETER = "ik-t"

imagekitio/url.py

@@ -11,19 +11,8 @@
 
 from .constants import ERRORS
 
-TRANSFORMATION_PARAMETER = "tr"
-DEFAULT_TRANSFORMATION_POSITION = "path"
-QUERY_TRANSFORMATION_POSITION = "query"
-CHAIN_TRANSFORM_DELIMITER = ":"
-TRANSFORM_DELIMITER = ","
-TRANSFORM_KEY_VALUE_DELIMITER = "-"
 
-SIGNATURE_PARAMETER = "ik-s"
-TIMESTAMP_PARAMETER = "ik-t"
-DEFAULT_TIMESTAMP = "9999999999"
-
-
-class Url(object):
+class Url:
     """
     Url class holds the request and related methods
     to generate url(signed and unsigned)
@@ -42,44 +31,47 @@ def build_url(self, options: dict) -> str:
         builds url for from all options,
         """
 
+        # important to strip the trailing slashes. later logic assumes no trailing slashes.
         path = options.get("path", "").strip("/")
         src = options.get("src", "").strip("/")
         url_endpoint = options.get("url_endpoint", "").strip("/")
         transformation_str = self.transformation_to_str(options.get("transformation"))
-        transformation_position = options.get("transformation_position") or DEFAULT_TRANSFORMATION_POSITION
+        transformation_position = options.get("transformation_position", Default.DEFAULT_TRANSFORMATION_POSITION.value)
 
         if transformation_position not in Default.VALID_TRANSFORMATION_POSITION.value:
             raise ValueError(ERRORS.INVALID_TRANSFORMATION_POSITION.value)
 
-        if (path is "" and src is ""):
+        if (path == "" and src == ""):
             return ""
 
-        if src:
-            temp_url = src.strip("/")
-            transformation_position = QUERY_TRANSFORMATION_POSITION
-        else:
+        # if path is present then it is given priority over src parameter
+        if path:
             if transformation_position == "path":
                 temp_url = "{}/{}:{}/{}".format(
-                    url_endpoint.strip("/"),
-                    TRANSFORMATION_PARAMETER,
+                    url_endpoint,
+                    Default.TRANSFORMATION_PARAMETER.value,
                     transformation_str.strip("/"),
-                    path.strip("/")
+                    path
                 )
             else:
                 temp_url = "{}/{}".format(
-                    url_endpoint.strip("/"),
-                    path.strip("/")
+                    url_endpoint,
+                    path
                 )
+        else:
+            temp_url = src
+            # if src parameter is used, then we force transformation position in query
+            transformation_position = Default.QUERY_TRANSFORMATION_POSITION.value
 
-        url_object = urlparse(temp_url.strip("/"))
+        url_object = urlparse(temp_url)
 
         query_params = dict(parse_qsl(url_object.query))
         query_params.update(options.get("query_parameters", {}))
-        if transformation_position == QUERY_TRANSFORMATION_POSITION:
-            query_params.update({"tr": transformation_str})
-        query_params.update({"ik-sdk-version": Default.SDK_VERSION.value})
+        if transformation_position == Default.QUERY_TRANSFORMATION_POSITION.value:
+            query_params.update({Default.TRANSFORMATION_PARAMETER.value: transformation_str})
+        query_params.update({Default.SDK_VERSION_PARAMETER.value: Default.SDK_VERSION.value})
 
-        # Update query params
+        # Update query params in the url
         url_object = url_object._replace(query=urlencode(query_params))
 
         if options.get("signed"):
@@ -92,39 +84,41 @@ def build_url(self, options: dict) -> str:
                 url_endpoint=url_endpoint,
                 expiry_timestamp=expiry_timestamp,
             )
-            query_params.update({TIMESTAMP_PARAMETER: expiry_timestamp, SIGNATURE_PARAMETER: url_signature})
+
+            """
+            If the expire_seconds parameter is specified then the output URL contains
+            ik-t parameter (unix timestamp seconds when the URL expires) and 
+            the signature contains the timestamp for computation. 
+            
+            If not present, then no ik-t parameter and the value 9999999999 is used.
+            """
+            if expire_seconds:
+                query_params.update({Default.TIMESTAMP_PARAMETER.value: expiry_timestamp, Default.SIGNATURE_PARAMETER.value: url_signature})
+            else:
+                query_params.update({Default.SIGNATURE_PARAMETER.value: url_signature})
+            
             # Update signature related query params
             url_object = url_object._replace(query=urlencode(query_params))
 
         return url_object.geturl()
 
     @staticmethod
-    def get_signature_timestamp(seconds: int = None) -> int:
+    def get_signature_timestamp(expiry_seconds: int = None) -> int:
         """
-        this function returns either default time stamp
-        or current unix time and expiry seconds to get
-        signature time stamp
+        this function returns the signature timestamp to be used
+        with the generated url. 
+        If expiry_seconds is provided, it returns expiry_seconds added
+        to the current unix time, otherwise the default time stamp
+        is returned.
         """
-        if not seconds:
+        if not expiry_seconds:
             return Default.DEFAULT_TIMESTAMP.value
         current_timestamp = int(dt.now().strftime("%s"))
 
-        return current_timestamp + seconds
-
-    @staticmethod
-    def prepare_dict_for_unparse(url_dict: dict) -> dict:
-        """
-        remove and add required back slash of 'netloc' and 'path'
-        to parse it properly, urllib.parse.unparse() function can't
-        create url properly if path doesn't have '/' at the start
-        """
-        url_dict["netloc"] = url_dict["netloc"].rstrip("/")
-        url_dict["path"] = "/" + url_dict["path"].strip("/")
-
-        return url_dict
+        return current_timestamp + expiry_seconds
 
     @staticmethod
-    def get_signature(private_key, url, url_endpoint, expiry_timestamp) -> str:
+    def get_signature(private_key, url, url_endpoint, expiry_timestamp : int) -> str:
         """"
         create signature(hashed hex key) from
         private_key, url, url_endpoint and expiry_timestamp
@@ -133,8 +127,8 @@ def get_signature(private_key, url, url_endpoint, expiry_timestamp) -> str:
         if url_endpoint[-1] != '/':
             url_endpoint += '/'
 
-        if isinstance(expiry_timestamp, int) and expiry_timestamp < 1:
-            expiry_timestamp = DEFAULT_TIMESTAMP
+        if expiry_timestamp < 1:
+            expiry_timestamp = Default.DEFAULT_TIMESTAMP.value
 
         replaced_url = url.replace(url_endpoint, "") + str(expiry_timestamp)
 
@@ -187,12 +181,12 @@ def transformation_to_str(transformation):
                     parsed_transform_step.append(
                         "{}{}{}".format(
                             transform_key,
-                            TRANSFORM_KEY_VALUE_DELIMITER,
+                            Default.TRANSFORM_KEY_VALUE_DELIMITER.value,
                             transformation[i][key],
                         )
                     )
 
             parsed_transforms.append(
-                TRANSFORM_DELIMITER.join(parsed_transform_step))
+                Default.TRANSFORM_DELIMITER.value.join(parsed_transform_step))
 
-        return CHAIN_TRANSFORM_DELIMITER.join(parsed_transforms)
+        return Default.CHAIN_TRANSFORM_DELIMITER.value.join(parsed_transforms)

imagekitio/utils/formatter.py

@@ -31,6 +31,8 @@ def request_formatter(data: dict) -> dict:
 
 
 def camel_dict_to_snake_dict(data: dict) -> dict:
+    """Convert the keys of dictionary from camel case to snake case
+    """
     return {camel_to_snake(key): val for key, val in data.items()}
 
 

tests/test_generate_url.py

@@ -192,7 +192,20 @@ def test_generate_url_with_path_and_signed_in_proper_form(self):
         }
 
         url = self.client.url(options)
-        self.assertIn("ik-s", url)
+        self.assertIn(Default.SIGNATURE_PARAMETER.value, url)
+
+    def test_generate_url_signed_without_expiry_does_not_have_timestamp_parameter(self):
+        """
+        Check query params does not contain timestamp parameter if expire_seconds isn't specified.
+        """
+        options = {
+            "path": "/test-signed-url.jpg",
+            "signed": True,
+            "transformation": [{"width": 100}],
+        }
+
+        url = self.client.url(options)
+        self.assertNotIn(Default.TIMESTAMP_PARAMETER.value, url)
 
     def test_url_with_new_transformation_returns_as_it_is(self):
         options = {
@@ -370,6 +383,23 @@ def test_url_signed_with_expire_in_seconds(self):
         url = self.client.url(options)
         self.assertIn("ik-t", url)
 
+    def test_generate_url_with_path_and_src_uses_path(self):
+        """
+        In case when both path and src fields are provided, the `path` should be preferred
+        """
+        options = {
+            "path": "/default-image.jpg",
+            "src": "https://ik.imagekit.io/ldt7znpgpjs/test_YhNhoRxWt.jpg",
+            "transformation": [{"height": "300", "width": "400"}],
+        }
+        url = self.client.url(options)
+        self.assertEqual(
+            url,
+            "https://test-domain.com/test-endpoint/tr:h-300,w-400/default-image.jpg?ik-sdk-version={}".format(
+                Default.SDK_VERSION.value
+            ),
+        )
+
     def test_get_signature_with_100_expire_seconds(self):
         url = "https://test-domain.com/test-endpoint/tr:w-100/test-signed-url.png"
         signature = self.client.url_obj.get_signature(