main:Add Search function with single and multiple query params

.gitignore

@@ -42,4 +42,5 @@ pydevd*.log
 nodeLanguageServer/**
 nodeLanguageServer.*/**
 dist/**
-venv
+venv
+myenv

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

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
+    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)

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 = '<Your 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))