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/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 18f1d6d..f017820 100644 --- a/ShipthisAPI/shipthisapi.py +++ b/ShipthisAPI/shipthisapi.py @@ -1,7 +1,11 @@ -from typing import Dict, List 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 @@ -10,75 +14,65 @@ 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, "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) - result = fetched_response.json() - - if fetched_response.status_code == 200: - if result.get("success"): + try: + 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() + + 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_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: 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_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: 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, 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 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 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 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 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) - 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 \ No newline at end of file + 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 6b0a456..be1b7b4 100644 --- a/demo.py +++ b/demo.py @@ -1,7 +1,28 @@ 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="sea_shipment", params={"count": 2})) # count is the number of records you need +# 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 +# search_data = update_delete(collection_name="airport", object_id="653a610e67c2e9c2bea3d301") +# print(shipthisapi.delete_item(item_data=search_data)) + +# query_payload = { +# "search_query": "S", +# "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, +# } +# 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