20
20
21
21
SIGNATURE_PARAMETER = "ik-s"
22
22
TIMESTAMP_PARAMETER = "ik-t"
23
- DEFAULT_TIMESTAMP = "9999999999"
24
23
25
24
26
- class Url ( object ) :
25
+ class Url :
27
26
"""
28
27
Url class holds the request and related methods
29
28
to generate url(signed and unsigned)
@@ -42,44 +41,47 @@ def build_url(self, options: dict) -> str:
42
41
builds url for from all options,
43
42
"""
44
43
44
+ # important to strip the trailing slashes. later logic assumes no trailing slashes.
45
45
path = options .get ("path" , "" ).strip ("/" )
46
46
src = options .get ("src" , "" ).strip ("/" )
47
47
url_endpoint = options .get ("url_endpoint" , "" ).strip ("/" )
48
48
transformation_str = self .transformation_to_str (options .get ("transformation" ))
49
- transformation_position = options .get ("transformation_position" ) or DEFAULT_TRANSFORMATION_POSITION
49
+ transformation_position = options .get ("transformation_position" , DEFAULT_TRANSFORMATION_POSITION )
50
50
51
51
if transformation_position not in Default .VALID_TRANSFORMATION_POSITION .value :
52
52
raise ValueError (ERRORS .INVALID_TRANSFORMATION_POSITION .value )
53
53
54
- if (path is "" and src is "" ):
54
+ if (path == "" and src == "" ):
55
55
return ""
56
56
57
- if src :
58
- temp_url = src .strip ("/" )
59
- transformation_position = QUERY_TRANSFORMATION_POSITION
60
- else :
57
+ # if path is present then it is given priority over src parameter
58
+ if path :
61
59
if transformation_position == "path" :
62
60
temp_url = "{}/{}:{}/{}" .format (
63
- url_endpoint . strip ( "/" ) ,
61
+ url_endpoint ,
64
62
TRANSFORMATION_PARAMETER ,
65
63
transformation_str .strip ("/" ),
66
- path . strip ( "/" )
64
+ path
67
65
)
68
66
else :
69
67
temp_url = "{}/{}" .format (
70
- url_endpoint . strip ( "/" ) ,
71
- path . strip ( "/" )
68
+ url_endpoint ,
69
+ path
72
70
)
71
+ else :
72
+ temp_url = src
73
+ # if src parameter is used, then we force transformation position in query
74
+ transformation_position = QUERY_TRANSFORMATION_POSITION
73
75
74
- url_object = urlparse (temp_url . strip ( "/" ) )
76
+ url_object = urlparse (temp_url )
75
77
76
78
query_params = dict (parse_qsl (url_object .query ))
77
79
query_params .update (options .get ("query_parameters" , {}))
78
80
if transformation_position == QUERY_TRANSFORMATION_POSITION :
79
81
query_params .update ({"tr" : transformation_str })
80
82
query_params .update ({"ik-sdk-version" : Default .SDK_VERSION .value })
81
83
82
- # Update query params
84
+ # Update query params in the url
83
85
url_object = url_object ._replace (query = urlencode (query_params ))
84
86
85
87
if options .get ("signed" ):
@@ -92,39 +94,41 @@ def build_url(self, options: dict) -> str:
92
94
url_endpoint = url_endpoint ,
93
95
expiry_timestamp = expiry_timestamp ,
94
96
)
95
- query_params .update ({TIMESTAMP_PARAMETER : expiry_timestamp , SIGNATURE_PARAMETER : url_signature })
97
+
98
+ """
99
+ If the expire_seconds parameter is specified then the output URL contains
100
+ ik-t parameter (unix timestamp seconds when the URL expires) and
101
+ the signature contains the timestamp for computation.
102
+
103
+ If not present, then no ik-t parameter and the value 9999999999 is used.
104
+ """
105
+ if expire_seconds :
106
+ query_params .update ({TIMESTAMP_PARAMETER : expiry_timestamp , SIGNATURE_PARAMETER : url_signature })
107
+ else :
108
+ query_params .update ({SIGNATURE_PARAMETER : url_signature })
109
+
96
110
# Update signature related query params
97
111
url_object = url_object ._replace (query = urlencode (query_params ))
98
112
99
113
return url_object .geturl ()
100
114
101
115
@staticmethod
102
- def get_signature_timestamp (seconds : int = None ) -> int :
116
+ def get_signature_timestamp (expiry_seconds : int = None ) -> int :
103
117
"""
104
- this function returns either default time stamp
105
- or current unix time and expiry seconds to get
106
- signature time stamp
118
+ this function returns the signature timestamp to be used
119
+ with the generated url.
120
+ If expiry_seconds is provided, it returns expiry_seconds added
121
+ to the current unix time, otherwise the default time stamp
122
+ is returned.
107
123
"""
108
- if not seconds :
124
+ if not expiry_seconds :
109
125
return Default .DEFAULT_TIMESTAMP .value
110
126
current_timestamp = int (dt .now ().strftime ("%s" ))
111
127
112
- return current_timestamp + seconds
113
-
114
- @staticmethod
115
- def prepare_dict_for_unparse (url_dict : dict ) -> dict :
116
- """
117
- remove and add required back slash of 'netloc' and 'path'
118
- to parse it properly, urllib.parse.unparse() function can't
119
- create url properly if path doesn't have '/' at the start
120
- """
121
- url_dict ["netloc" ] = url_dict ["netloc" ].rstrip ("/" )
122
- url_dict ["path" ] = "/" + url_dict ["path" ].strip ("/" )
123
-
124
- return url_dict
128
+ return current_timestamp + expiry_seconds
125
129
126
130
@staticmethod
127
- def get_signature (private_key , url , url_endpoint , expiry_timestamp ) -> str :
131
+ def get_signature (private_key , url , url_endpoint , expiry_timestamp : int ) -> str :
128
132
""""
129
133
create signature(hashed hex key) from
130
134
private_key, url, url_endpoint and expiry_timestamp
@@ -133,8 +137,8 @@ def get_signature(private_key, url, url_endpoint, expiry_timestamp) -> str:
133
137
if url_endpoint [- 1 ] != '/' :
134
138
url_endpoint += '/'
135
139
136
- if isinstance ( expiry_timestamp , int ) and expiry_timestamp < 1 :
137
- expiry_timestamp = DEFAULT_TIMESTAMP
140
+ if expiry_timestamp < 1 :
141
+ expiry_timestamp = Default . DEFAULT_TIMESTAMP . value
138
142
139
143
replaced_url = url .replace (url_endpoint , "" ) + str (expiry_timestamp )
140
144
0 commit comments