From dd5efc0cd0c02833d9936ef2a4dfee1464eeb5c5 Mon Sep 17 00:00:00 2001 From: Junaid Date: Thu, 19 Oct 2023 17:34:24 +0530 Subject: [PATCH 1/5] main:Add Search function with single and multiple query params --- .gitignore | 3 ++- ShipthisAPI/shipthisapi.py | 41 +++++++++++++++++++++++++++++++++++--- demo.py | 22 +++++++++++++++++--- 3 files changed, 59 insertions(+), 7 deletions(-) diff --git a/.gitignore b/.gitignore index 5637e6e..04a3b00 100644 --- a/.gitignore +++ b/.gitignore @@ -42,4 +42,5 @@ pydevd*.log nodeLanguageServer/** nodeLanguageServer.*/** dist/** -venv \ No newline at end of file +venv +myenv \ No newline at end of file diff --git a/ShipthisAPI/shipthisapi.py b/ShipthisAPI/shipthisapi.py index 18f1d6d..3845bac 100644 --- a/ShipthisAPI/shipthisapi.py +++ b/ShipthisAPI/shipthisapi.py @@ -1,5 +1,7 @@ -from typing import Dict, List import requests +from typing import List, Dict, Union, Any +import json + class ShipthisAPI: base_api_endpoint = 'https://api.shipthis.co/api/v3/' @@ -22,8 +24,11 @@ def _make_request(self, method: str, path: str, query_params: str=None, request_ "user_type": self.user_type, "location": 'new_york' } - fetched_response = requests.request(method, self.base_api_endpoint + path, data=request_data or {}, headers=headers, params=query_params) + print(self.base_api_endpoint + path) + fetched_response = requests.request(method, self.base_api_endpoint + path, data=request_data or {}, headers=headers) result = fetched_response.json() + + print(result) if fetched_response.status_code == 200: if result.get("success"): @@ -81,4 +86,34 @@ def delete_item(self, collection_name: str, object_id: str) -> Dict: # if isinstance(resp, str): # return resp # else: - return resp \ No newline at end of file + return resp + + def get_search_list_collection(self, collection_name: str, query_filter: str, params= None) -> Dict: + path = f'incollection/{collection_name}?search_query={query_filter}' + get_list_response = self._make_request('GET', path) + if isinstance(get_list_response, str): + return get_list_response + else: + if get_list_response.get("items", False): + return get_list_response.get("items", []) + else: + return "No items found for the provided query." + + def get_full_search_list_collection(self, collection_name: str, query_params: Dict[str, any], params= None) ->Dict: + strr = '' + for key in query_params: + valueQuery = query_params[key] + if not type(query_params[key]) is str: + valueQuery = json.dumps(query_params[key]) + + if(len(strr)==0): + strr += '?' + key + '=' + valueQuery + else: + strr += '&' + key + '=' + valueQuery + + path = f'incollection/{collection_name}{strr}' + response = self._make_request('GET', path) + if isinstance(response, str): + return response + else: + return response diff --git a/demo.py b/demo.py index 6b0a456..2504e47 100644 --- a/demo.py +++ b/demo.py @@ -1,7 +1,23 @@ from ShipthisAPI.shipthisapi import ShipthisAPI -x_api_key = '' +x_api_key = 'WyJqdW5haWRAc2hpcHRoaXMuY28iLCJkZW1vIl0.ZS0mOA.k2yBEld1jx7perTOqgLjYIELUkw' shipthisapi = ShipthisAPI(organisation='demo', x_api_key=x_api_key, region_id='usa', location_id='new_york') -print(shipthisapi.get_list(collection_name="invoice")) -print(shipthisapi.get_list(collection_name="sea_shipment", params={"count": 2})) # count is the number of records you need +# print(shipthisapi.get_list(collection_name="invoice")) +# print(shipthisapi.get_list(collection_name="sea_shipment", params={"count": 2})) # count is the number of records you need +# print(shipthisapi.get_search_list_collection(collection_name="airport",query_filter="Salekhard")) +query_payload = { + "search_query": "Salekhard", + "count": 25, + "page": 1, + "multi_sort": [{"sort_by": "created_at", "sort_order": "dsc"}], + "output_type": "json", + "meta": False, + "queryFilterV2": [], + "general_filter": {"job_status": {"$nin": ["closed", "cancelled", "ops_complete"]}}, + "only": "job_id,shipment_name,shipment_status,customer_name.company.name,customer_name._id,customer_name._cls,customer_name._id,consignee_name.company.name,consignee_name._id,shipper_name.company.name,shipper_name._id,mawb_no,hawb_no", + "location": "new_york", + "region_override": False, +} + +print(shipthisapi.get_full_search_list_collection(collection_name="airport",query_params=query_payload)) \ No newline at end of file From df194619b42495374aa3e3f04cba43a2aba702d2 Mon Sep 17 00:00:00 2001 From: Junaid Date: Thu, 19 Oct 2023 17:41:50 +0530 Subject: [PATCH 2/5] main:Removing Api key and other testing fileds --- demo.py | 30 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/demo.py b/demo.py index 2504e47..7cf177d 100644 --- a/demo.py +++ b/demo.py @@ -1,23 +1,23 @@ from ShipthisAPI.shipthisapi import ShipthisAPI -x_api_key = 'WyJqdW5haWRAc2hpcHRoaXMuY28iLCJkZW1vIl0.ZS0mOA.k2yBEld1jx7perTOqgLjYIELUkw' +x_api_key = '' shipthisapi = ShipthisAPI(organisation='demo', x_api_key=x_api_key, region_id='usa', location_id='new_york') # print(shipthisapi.get_list(collection_name="invoice")) # print(shipthisapi.get_list(collection_name="sea_shipment", params={"count": 2})) # count is the number of records you need # print(shipthisapi.get_search_list_collection(collection_name="airport",query_filter="Salekhard")) -query_payload = { - "search_query": "Salekhard", - "count": 25, - "page": 1, - "multi_sort": [{"sort_by": "created_at", "sort_order": "dsc"}], - "output_type": "json", - "meta": False, - "queryFilterV2": [], - "general_filter": {"job_status": {"$nin": ["closed", "cancelled", "ops_complete"]}}, - "only": "job_id,shipment_name,shipment_status,customer_name.company.name,customer_name._id,customer_name._cls,customer_name._id,consignee_name.company.name,consignee_name._id,shipper_name.company.name,shipper_name._id,mawb_no,hawb_no", - "location": "new_york", - "region_override": False, -} +# query_payload = { +# "search_query": "Salekhard", +# "count": 25, +# "page": 1, +# "multi_sort": [{"sort_by": "created_at", "sort_order": "dsc"}], +# "output_type": "json", +# "meta": False, +# "queryFilterV2": [], +# "general_filter": {"job_status": {"$nin": ["closed", "cancelled", "ops_complete"]}}, +# "only": "job_id,shipment_name,shipment_status,customer_name.company.name,customer_name._id,customer_name._cls,customer_name._id,consignee_name.company.name,consignee_name._id,shipper_name.company.name,shipper_name._id,mawb_no,hawb_no", +# "location": "new_york", +# "region_override": False, +# } -print(shipthisapi.get_full_search_list_collection(collection_name="airport",query_params=query_payload)) \ No newline at end of file +# print(shipthisapi.get_full_search_list_collection(collection_name="airport",query_params=query_payload)) \ No newline at end of file From ade4b9aecfd73cae62ad3e58617cfc8abcff7110 Mon Sep 17 00:00:00 2001 From: Junaid Date: Mon, 23 Oct 2023 16:44:06 +0530 Subject: [PATCH 3/5] feature/search: removing print line & demo edit --- ShipthisAPI/shipthisapi.py | 13 +++++-------- demo.py | 7 ++++++- 2 files changed, 11 insertions(+), 9 deletions(-) diff --git a/ShipthisAPI/shipthisapi.py b/ShipthisAPI/shipthisapi.py index 3845bac..eb5b591 100644 --- a/ShipthisAPI/shipthisapi.py +++ b/ShipthisAPI/shipthisapi.py @@ -24,12 +24,9 @@ def _make_request(self, method: str, path: str, query_params: str=None, request_ "user_type": self.user_type, "location": 'new_york' } - print(self.base_api_endpoint + path) fetched_response = requests.request(method, self.base_api_endpoint + path, data=request_data or {}, headers=headers) result = fetched_response.json() - print(result) - if fetched_response.status_code == 200: if result.get("success"): return result.get("data") @@ -100,18 +97,18 @@ def get_search_list_collection(self, collection_name: str, query_filter: str, pa return "No items found for the provided query." def get_full_search_list_collection(self, collection_name: str, query_params: Dict[str, any], params= None) ->Dict: - strr = '' + query_path = '' for key in query_params: valueQuery = query_params[key] if not type(query_params[key]) is str: valueQuery = json.dumps(query_params[key]) - if(len(strr)==0): - strr += '?' + key + '=' + valueQuery + if(len(query_path)==0): + query_path += '?' + key + '=' + valueQuery else: - strr += '&' + key + '=' + valueQuery + query_path += '&' + key + '=' + valueQuery - path = f'incollection/{collection_name}{strr}' + path = f'incollection/{collection_name}{query_path}' response = self._make_request('GET', path) if isinstance(response, str): return response diff --git a/demo.py b/demo.py index 7cf177d..7342d0f 100644 --- a/demo.py +++ b/demo.py @@ -5,7 +5,12 @@ # print(shipthisapi.get_list(collection_name="invoice")) # print(shipthisapi.get_list(collection_name="sea_shipment", params={"count": 2})) # count is the number of records you need -# print(shipthisapi.get_search_list_collection(collection_name="airport",query_filter="Salekhard")) + + +# Fucntion call for getting full search list with query parameteres + +# print(shipthisapi.get_search_list_collection(collection_name="airport",query_filter="")) + # query_payload = { # "search_query": "Salekhard", # "count": 25, From 1eeb2bac44002259527b4224e6e94b9308a43bc5 Mon Sep 17 00:00:00 2001 From: Junaid Date: Mon, 23 Oct 2023 17:48:14 +0530 Subject: [PATCH 4/5] feature/search:error handeling for api response --- ShipthisAPI/shipthisapi.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/ShipthisAPI/shipthisapi.py b/ShipthisAPI/shipthisapi.py index eb5b591..33eed9d 100644 --- a/ShipthisAPI/shipthisapi.py +++ b/ShipthisAPI/shipthisapi.py @@ -24,8 +24,16 @@ def _make_request(self, method: str, path: str, query_params: str=None, request_ "user_type": self.user_type, "location": 'new_york' } - fetched_response = requests.request(method, self.base_api_endpoint + path, data=request_data or {}, headers=headers) - result = fetched_response.json() + try: + fetched_response = requests.request(method, self.base_api_endpoint + path, data=request_data or {}, headers=headers, params=query_params) + fetched_response.raise_for_status() #Raise an exception for HTTP errors + result = fetched_response.json() + except requests.exceptions.RequestException as e: + print(f"An error occurred while making the API request: {e}") + except ValueError as e: + print(f"An error occurred while parsing the JSON response: {e}") + except Exception as e: + print(f"An unexpected error occurred: {e}") if fetched_response.status_code == 200: if result.get("success"): From ed274757f22ac3e9a9e70515c6bd230405215935 Mon Sep 17 00:00:00 2001 From: Junaid Date: Thu, 26 Oct 2023 18:52:29 +0530 Subject: [PATCH 5/5] feature/search:Added Pydantic to all the API calls --- ShipthisAPI/getCollection.py | 32 +++++++++ ShipthisAPI/shipthisapi.py | 126 +++++++++++------------------------ demo.py | 18 ++--- 3 files changed, 81 insertions(+), 95 deletions(-) create mode 100644 ShipthisAPI/getCollection.py diff --git a/ShipthisAPI/getCollection.py b/ShipthisAPI/getCollection.py new file mode 100644 index 0000000..f546485 --- /dev/null +++ b/ShipthisAPI/getCollection.py @@ -0,0 +1,32 @@ +from pydantic import BaseModel, validator, constr +from typing import Any + +class get_Collection(BaseModel): + collection_name: constr(min_length=3) + +class update_delete(BaseModel): + collection_name: str + object_id: constr(min_length=3) + +class get_search_list_collection(BaseModel): + collection_name: str + query_filter: str + + @validator('query_filter') + def check_query_filter_length(cls, query_filter): + if len(query_filter) == 0: + raise ValueError("Query filter cannot be empty.") + return query_filter + +class get_full_search_list_collection(BaseModel): + collection_name: str + query_params: dict[str, Any] + + @validator('query_params', pre=True, each_item=True) + def check_query_params_structure(cls, query_params): + if not isinstance(query_params, dict): + raise ValueError("Query parameters should be of type dictionary.") + for key, value in query_params.items(): + if not isinstance(key, str): + raise ValueError(f"Key '{key}' is not a string.") + return query_params diff --git a/ShipthisAPI/shipthisapi.py b/ShipthisAPI/shipthisapi.py index 33eed9d..f017820 100644 --- a/ShipthisAPI/shipthisapi.py +++ b/ShipthisAPI/shipthisapi.py @@ -1,9 +1,11 @@ import requests from typing import List, Dict, Union, Any import json +from .getCollection import get_Collection, update_delete, get_search_list_collection, get_full_search_list_collection + class ShipthisAPI: - base_api_endpoint = 'https://api.shipthis.co/api/v3/' + BASE_API_ENDPOINT = 'https://api.shipthis.co/api/v3/' def __init__(self, organisation: str, x_api_key:str, user_type='employee', region_id: str=None, location_id: str=None) -> None: self.x_api_key = x_api_key @@ -12,12 +14,11 @@ def __init__(self, organisation: str, x_api_key:str, user_type='employee', regio self.region_id = region_id self.location_id = location_id - def set_region_location(self, region_id, location_id): self.region_id = region_id self.location_id = location_id - - def _make_request(self, method: str, path: str, query_params: str=None, request_data=None) -> None: + + def _make_request(self, method: str, path: str, query_params: Dict=None, request_data: Dict=None) -> Union[Dict, str]: headers = { "x-api-key": self.x_api_key, "organisation": self.organisation_id, @@ -25,100 +26,53 @@ def _make_request(self, method: str, path: str, query_params: str=None, request_ "location": 'new_york' } try: - fetched_response = requests.request(method, self.base_api_endpoint + path, data=request_data or {}, headers=headers, params=query_params) - fetched_response.raise_for_status() #Raise an exception for HTTP errors + fetched_response = requests.request(method, self.BASE_API_ENDPOINT + path, json=request_data, headers=headers, params=query_params) + fetched_response.raise_for_status() result = fetched_response.json() - except requests.exceptions.RequestException as e: - print(f"An error occurred while making the API request: {e}") - except ValueError as e: - print(f"An error occurred while parsing the JSON response: {e}") - except Exception as e: - print(f"An unexpected error occurred: {e}") - if fetched_response.status_code == 200: - if result.get("success"): + if fetched_response.status_code == 200 and result.get("success", True): + if (result.get('success')): + return "Successfully completed the request" + # if (result.get('data', True)): + # return "Please provide the proper information to get all the desired results" return result.get("data") else: error_message = result.get("errors") or "API call failed. Please check your internet connection or try again later" return error_message[0].get('message') if error_message[0].get('message') else "Please provide the necessary requirements or try again later" - else: - return "Internal Server error, please try again later" + except (requests.exceptions.RequestException, ValueError, Exception) as e: + return f"An error occurred: {e}" def info(self) -> Dict: - info_resp = self._make_request('GET', 'auth/info') - return info_resp - - + return self._make_request('GET', 'auth/info') - def get_one_item(self, collection_name: str, params=None) -> Dict: - resp = self._make_request('GET', 'incollection/' + collection_name) - if isinstance(resp, dict): - if resp.get("items"): - # return first elem - return resp.get("items")[0] - else: - return resp - - def get_list(self, collection_name: str, params=None) -> List[Dict] or str: - get_list_response = self._make_request('GET', 'incollection/' + collection_name, params) - if isinstance(get_list_response, str): - return get_list_response - else: - if get_list_response.get("items", False): - return get_list_response.get("items", []) - else: - return get_list_response + def get_one_item(self, collection_name: get_Collection) -> Dict: + resp = self._make_request('GET', f'incollection/{collection_name.collection_name}') + if isinstance(resp, dict) and resp.get("items"): + return resp.get("items")[0] + return resp + def get_list(self, collection_name: get_Collection,params=None) -> Union[List[Dict], str]: + response = self._make_request('GET', f'incollection/{collection_name}',params) + if isinstance(response, dict) and response.get("items"): + return response.get("items") + return response - def create_item(self, collection_name: str, data=None) -> Dict: - resp = self._make_request('POST', 'incollection/' + collection_name, request_data={"reqbody": data}) - if isinstance(resp, dict): - if resp.get("data"): - return resp.get("data") - else: - return resp + def create_item(self, collection_name: get_Collection, data: Dict) -> Dict: + resp = self._make_request('POST', f'incollection/{collection_name.collection_name}', request_data={"reqbody": data}) + return resp if isinstance(resp, dict) and resp.get("data") else resp - def update_item(self, collection_name: str, object_id: str, updated_data=None) -> Dict: - resp = self._make_request('PUT', 'incollection/' + collection_name + '/' + object_id, request_data={"reqbody": updated_data}) - if isinstance(resp, dict): - if resp.get("data"): - return resp.get("data") - else: - return resp + def update_item(self, item_data: update_delete, updated_data: Dict=None) -> Dict: + resp = self._make_request('PUT', f'incollection/{item_data.collection_name}/{str(item_data.object_id)}', request_data={"reqbody": updated_data}) + return resp if isinstance(resp, dict) and resp.get("data") else resp - def delete_item(self, collection_name: str, object_id: str) -> Dict: - resp = self._make_request('DELETE', 'incollection/' + collection_name + '/' + object_id) - # if isinstance(resp, str): - # return resp - # else: - return resp + def delete_item(self, item_data: update_delete) -> Dict: + return self._make_request('DELETE', f'incollection/{item_data.collection_name}/{str(item_data.object_id)}') - def get_search_list_collection(self, collection_name: str, query_filter: str, params= None) -> Dict: - path = f'incollection/{collection_name}?search_query={query_filter}' - get_list_response = self._make_request('GET', path) - if isinstance(get_list_response, str): - return get_list_response - else: - if get_list_response.get("items", False): - return get_list_response.get("items", []) - else: - return "No items found for the provided query." - - def get_full_search_list_collection(self, collection_name: str, query_params: Dict[str, any], params= None) ->Dict: - query_path = '' - for key in query_params: - valueQuery = query_params[key] - if not type(query_params[key]) is str: - valueQuery = json.dumps(query_params[key]) + def get_search_list_collection(self, item_data: get_search_list_collection) -> Dict: + path = f'incollection/{item_data.collection_name}?search_query={item_data.query_filter}' + return self._make_request('GET', path) - if(len(query_path)==0): - query_path += '?' + key + '=' + valueQuery - else: - query_path += '&' + key + '=' + valueQuery - - path = f'incollection/{collection_name}{query_path}' - response = self._make_request('GET', path) - if isinstance(response, str): - return response - else: - return response + def get_full_search_list_collection(self, item_data: get_full_search_list_collection) ->Dict: + query_path = '&'.join([f"{key}={json.dumps(value) if not isinstance(value, str) else value}" for key, value in item_data.query_params.items()]) + path = f'incollection/{item_data.collection_name}?{query_path}' + return self._make_request('GET', path) diff --git a/demo.py b/demo.py index 7342d0f..be1b7b4 100644 --- a/demo.py +++ b/demo.py @@ -1,18 +1,18 @@ from ShipthisAPI.shipthisapi import ShipthisAPI - -x_api_key = '' +from ShipthisAPI.getCollection import get_Collection, update_delete, get_search_list_collection, get_full_search_list_collection +x_api_key = '' shipthisapi = ShipthisAPI(organisation='demo', x_api_key=x_api_key, region_id='usa', location_id='new_york') -# print(shipthisapi.get_list(collection_name="invoice")) +# print(shipthisapi.get_list(collection_name="airport")) # print(shipthisapi.get_list(collection_name="sea_shipment", params={"count": 2})) # count is the number of records you need -# Fucntion call for getting full search list with query parameteres - -# print(shipthisapi.get_search_list_collection(collection_name="airport",query_filter="")) +# # Fucntion call for getting full search list with query parameteres +# search_data = update_delete(collection_name="airport", object_id="653a610e67c2e9c2bea3d301") +# print(shipthisapi.delete_item(item_data=search_data)) # query_payload = { -# "search_query": "Salekhard", +# "search_query": "S", # "count": 25, # "page": 1, # "multi_sort": [{"sort_by": "created_at", "sort_order": "dsc"}], @@ -24,5 +24,5 @@ # "location": "new_york", # "region_override": False, # } - -# print(shipthisapi.get_full_search_list_collection(collection_name="airport",query_params=query_payload)) \ No newline at end of file +# search_data = get_full_search_list_collection(collection_name="airport",query_params=query_payload) +# print(shipthisapi.get_full_search_list_collection(item_data=search_data)) \ No newline at end of file