Skip to content

Commit 681677b

Browse files
author
Samiul Sk
committed
feat: add file related functionalities
1 parent bf62a8c commit 681677b

1 file changed

Lines changed: 246 additions & 0 deletions

File tree

‎imagekitio/file.py‎

Lines changed: 246 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,246 @@
1+
import json
2+
from typing import Any, Dict
3+
4+
from .constants.errors import ERRORS
5+
from .constants.files import VALID_FILE_OPTIONS, VALID_UPLOAD_OPTIONS
6+
from .constants.url import URL
7+
from .utils.formatter import (
8+
camel_dict_to_snake_dict,
9+
request_formatter,
10+
snake_to_lower_camel,
11+
)
12+
13+
14+
class File(object):
15+
def __init__(self, request_obj):
16+
self.request = request_obj
17+
18+
def upload(self, file, file_name, options) -> Dict:
19+
"""Upload file to server using local image or url
20+
:param file: either local file path or network file path
21+
:param file_name: intended file name
22+
:param options: intended options
23+
:return: json response from server
24+
"""
25+
if not file:
26+
raise TypeError(ERRORS.MISSING_UPLOAD_FILE_PARAMETER.value)
27+
if not file_name:
28+
raise TypeError(ERRORS.MISSING_UPLOAD_FILENAME_PARAMETER.value)
29+
url = URL.BASE_URL.value + URL.UPLOAD.value
30+
headers = self.request.create_headers()
31+
32+
files = {
33+
"file": file,
34+
"fileName": (None, file_name),
35+
}
36+
37+
if not options:
38+
options = dict()
39+
else:
40+
options = self.validate_upload(options)
41+
if options is False:
42+
raise ValueError("Invalid upload options")
43+
if isinstance(file, str) or isinstance(file, bytes):
44+
files.update({"file": (None, file)})
45+
resp = self.request.request(
46+
"Post", url=url, files=files, data=options, headers=headers
47+
)
48+
49+
if resp.status_code > 200:
50+
error = resp.json()
51+
response = None
52+
else:
53+
error = None
54+
response = resp.json()
55+
response = {"error": error, "response": response}
56+
return response
57+
58+
def list(self, options: dict) -> Dict:
59+
"""Returns list files on ImageKit Server
60+
:param: options dictionary of options
61+
:return: list of the response
62+
"""
63+
64+
formatted_options = request_formatter(options)
65+
if not self.is_valid_list_options(formatted_options):
66+
raise ValueError("Invalid option for list_files")
67+
url = URL.BASE_URL.value
68+
headers = self.request.create_headers()
69+
70+
resp = self.request.request(
71+
method="GET", url=url, headers=headers, params=options
72+
)
73+
if resp.status_code > 200:
74+
error = resp.json()
75+
response = None
76+
else:
77+
error = None
78+
response = resp.json()
79+
response = {"error": error, "response": response}
80+
return response
81+
82+
def details(self, file_identifier: str = None) -> Dict:
83+
"""returns file detail
84+
"""
85+
if not file_identifier:
86+
raise TypeError(ERRORS.FILE_ID_MISSING.value)
87+
url = "{}/{}/details".format(URL.BASE_URL.value, file_identifier)
88+
resp = self.request.request(
89+
method="GET", url=url, headers=self.request.create_headers(),
90+
)
91+
if resp.status_code > 200:
92+
error = resp.json()
93+
response = None
94+
else:
95+
error = None
96+
response = resp.json()
97+
response = {"error": error, "response": response}
98+
return response
99+
100+
def update_file_details(self, file_id: str, options: dict):
101+
"""Update detail of a file(like tags, coordinates)
102+
update details identified by file_id and options,
103+
which is already uploaded
104+
"""
105+
106+
if not ("tags" in options.keys() or "custom_coordinates" in options.keys()):
107+
raise ValueError(ERRORS.UPDATE_DATA_MISSING.value)
108+
if not isinstance(options.get("tags", []), list):
109+
raise ValueError(ERRORS.UPDATE_DATA_TAGS_INVALID.value)
110+
if not isinstance(options.get("custom_coordinates", ""), str):
111+
raise ValueError(ERRORS.UPDATE_DATA_COORDS_INVALID.value)
112+
url = "{}/{}/details/".format(URL.BASE_URL.value, file_id)
113+
headers = {"Content-Type": "application/json"}
114+
headers.update(self.request.get_auth_headers())
115+
data = json.dumps(request_formatter(options))
116+
resp = self.request.request(method="Patch", url=url, headers=headers, data=data)
117+
if resp.status_code > 200:
118+
error = resp.json()
119+
response = None
120+
else:
121+
error = None
122+
response = resp.json()
123+
response = {"error": error, "response": response}
124+
return response
125+
126+
def delete(self, file_id: str = None) -> Dict:
127+
"""Delete file by file_id
128+
deletes file from imagekit server
129+
"""
130+
if not file_id:
131+
raise TypeError(ERRORS.FILE_ID_MISSING.value)
132+
url = "{}/{}".format(URL.BASE_URL.value, file_id)
133+
resp = self.request.request(
134+
method="Delete", url=url, headers=self.request.create_headers()
135+
)
136+
if resp.status_code > 204:
137+
error = resp.text
138+
response = None
139+
else:
140+
error = None
141+
response = None
142+
response = {"error": error, "response": response}
143+
return response
144+
145+
def purge_cache(self, file_url: str = None) -> Dict[str, Any]:
146+
"""Use from child class to purge cache
147+
"""
148+
if not file_url:
149+
raise TypeError(ERRORS.MISSING_FILE_URL.value)
150+
url = URL.BASE_URL.value + "/purge"
151+
headers = {"Content-Type": "application/json"}
152+
headers.update(self.request.get_auth_headers())
153+
body = {"url": file_url}
154+
resp = self.request.request(
155+
"Post", headers=headers, url=url, data=json.dumps(body)
156+
)
157+
formatted_resp = camel_dict_to_snake_dict(resp.json())
158+
if resp.status_code > 204:
159+
error = formatted_resp
160+
response = None
161+
else:
162+
error = None
163+
response = formatted_resp
164+
response = {"error": error, "response": response}
165+
return response
166+
167+
def get_purge_cache_status(self, cache_request_id: str = None) -> Dict[str, Any]:
168+
"""Get purge cache status by cache_request_id
169+
:return: cache_request_id
170+
"""
171+
if not cache_request_id:
172+
raise TypeError(ERRORS.CACHE_PURGE_STATUS_ID_MISSING.value)
173+
174+
url = "{}/purge/{}".format(URL.BASE_URL.value, cache_request_id)
175+
headers = self.request.create_headers()
176+
resp = self.request.request("GET", url, headers=headers)
177+
formatted_resp = camel_dict_to_snake_dict(resp.json())
178+
179+
if resp.status_code > 200:
180+
error = formatted_resp
181+
response = None
182+
else:
183+
error = None
184+
response = formatted_resp
185+
response = {"error": error, "response": response}
186+
return response
187+
188+
def get_metadata(self, file_id: str = None):
189+
"""Get metadata by file_id
190+
"""
191+
if not file_id:
192+
raise TypeError(ERRORS.FILE_ID_MISSING.value)
193+
194+
url = "{}/{}/metadata".format(URL.BASE_URL.value, file_id)
195+
resp = self.request.request("GET", url, headers=self.request.create_headers())
196+
formatted_resp = camel_dict_to_snake_dict(resp.json())
197+
if resp.status_code > 200:
198+
error = formatted_resp
199+
response = None
200+
else:
201+
error = None
202+
response = formatted_resp
203+
response = {"error": error, "response": response}
204+
return response
205+
206+
def is_valid_list_options(self, options: Dict[str, Any]) -> bool:
207+
"""Returns if options are valid
208+
"""
209+
valid_values = self.get_valid_list_values()
210+
for key in options:
211+
if key not in valid_values:
212+
return False
213+
return True
214+
215+
@staticmethod
216+
def get_valid_list_values():
217+
"""Returns valid options for list files
218+
"""
219+
return VALID_FILE_OPTIONS
220+
221+
@staticmethod
222+
def validate_upload(options):
223+
"""
224+
Validates upload value, checks if params are valid,
225+
changes snake to camel case
226+
"""
227+
response_list = []
228+
for key, val in options.items():
229+
if key not in VALID_UPLOAD_OPTIONS:
230+
return False
231+
if key == "response_fields":
232+
for i, j in enumerate(options[key]):
233+
if j not in VALID_UPLOAD_OPTIONS:
234+
return False
235+
response_list.append(snake_to_lower_camel(j))
236+
options[key] = ",".join(response_list)
237+
continue
238+
if isinstance(val, list):
239+
val = ",".join(val)
240+
options[key] = val
241+
continue
242+
# imagekit server accepts 'true/false'
243+
elif isinstance(val, bool):
244+
val = str(val).lower()
245+
options[key] = val
246+
return request_formatter(options)

0 commit comments

Comments
 (0)
⚡