Coverage for mindsdb / integrations / handlers / bigcommerce_handler / bigcommerce_api_client.py: 0%
134 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-21 00:36 +0000
1from http import HTTPStatus
2from urllib.parse import urljoin
4import requests
7DEFAULT_LIMIT = 999999999
10class BigCommerceAPIClient:
11 def __init__(self, url: str, access_token: str):
12 # we have to use both endpoints: v2/ and v3/, so delete it from base url
13 self.base_url = url.rstrip("/")
14 self.base_url = self.base_url.rstrip("v2")
15 self.base_url = self.base_url.rstrip("v3")
16 if not self.base_url.endswith("/"):
17 self.base_url += "/"
19 self.access_token = access_token
20 self.session = requests.Session()
21 self.session.headers.update(
22 {
23 "X-Auth-Token": self.access_token,
24 "Content-Type": "application/json",
25 "Accept": "application/json",
26 }
27 )
29 def get_products(
30 self,
31 filter: dict = None,
32 sort_condition: dict = None,
33 limit: int = None,
34 ):
35 # doc: https://developer.bigcommerce.com/docs/rest-catalog/products#get-all-products
36 params = {
37 "limit": limit or DEFAULT_LIMIT,
38 }
39 if filter is not None:
40 params.update(filter)
41 if sort_condition is not None:
42 params.update(sort_condition)
43 return self._make_request_v3("GET", "catalog/products", params=params)
45 def get_customers(
46 self,
47 filter: dict = None,
48 sort_condition: dict = None,
49 limit: int = None,
50 ):
51 # doc: https://developer.bigcommerce.com/docs/rest-management/customers#get-all-customers
52 params = {
53 "limit": limit or DEFAULT_LIMIT,
54 }
55 if filter:
56 params.update(filter)
57 if sort_condition:
58 params["sort"] = sort_condition
59 return self._make_request_v3("GET", "customers", params=params)
61 def get_orders(
62 self,
63 filter: dict = None,
64 sort_condition: str = None,
65 limit: int = None,
66 ):
67 # doc: https://developer.bigcommerce.com/docs/rest-management/orders#get-all-orders
68 params = {"limit": limit or DEFAULT_LIMIT}
69 if filter:
70 params.update(filter)
71 if sort_condition:
72 params["sort"] = sort_condition
74 return self._make_request_v2("GET", "orders", params=params)
76 def get_orders_count(self) -> int:
77 response = self._make_request_v2("GET", "orders/count")
78 return response["count"]
80 def get_customers_count(self) -> int:
81 response = self._make_request("GET", urljoin(self.base_url, "v3/customers"), params={"limit": 1})
82 return response["meta"]["pagination"]["total"]
84 def get_products_count(self) -> int:
85 response = self._make_request("GET", urljoin(self.base_url, "v3/products"), params={"limit": 1})
86 return response["meta"]["pagination"]["total"]
88 def get_categories(
89 self,
90 filter: dict = None,
91 limit: int = None,
92 ):
93 # doc: https://developer.bigcommerce.com/docs/rest-catalog/category-trees/categories#get-all-categories
94 params = {
95 "limit": limit or DEFAULT_LIMIT,
96 }
97 if filter is not None:
98 params.update(filter)
99 return self._make_request_v3("GET", "catalog/trees/categories", params=params)
101 def get_categories_count(self) -> int:
102 response = self._make_request("GET", urljoin(self.base_url, "v3/catalog/trees/categories"), params={"limit": 1})
103 return response.get("meta", {}).get("pagination", {}).get("total", 0)
105 def get_pickups(
106 self,
107 filter: dict = None,
108 limit: int = None,
109 ):
110 # doc: https://developer.bigcommerce.com/docs/rest-management/pickup#get-pickups
111 params = {
112 "limit": limit or DEFAULT_LIMIT,
113 }
114 if filter is not None:
115 params.update(filter)
116 return self._make_request_v3("GET", "orders/pickups", params=params)
118 def get_pickups_count(self) -> int:
119 response = self._make_request("GET", urljoin(self.base_url, "v3/pickups"), params={"limit": 1})
120 return response.get("meta", {}).get("pagination", {}).get("total", 0)
122 def get_promotions(
123 self,
124 filter: dict = None,
125 sort_condition: dict = None,
126 limit: int = None,
127 ):
128 # doc: https://developer.bigcommerce.com/docs/rest-management/promotions/promotions-bulk#get-all-promotions
129 params = {
130 "limit": limit or DEFAULT_LIMIT,
131 }
132 if filter is not None:
133 params.update(filter)
134 if sort_condition is not None:
135 params.update(sort_condition)
136 return self._make_request_v3("GET", "promotions", params=params)
138 def get_promotions_count(self) -> int:
139 response = self._make_request("GET", urljoin(self.base_url, "v3/promotions"), params={"limit": 1})
140 return response.get("meta", {}).get("pagination", {}).get("total", 0)
142 def get_wishlists(
143 self,
144 filter: dict = None,
145 limit: int = None,
146 ):
147 # doc: https://developer.bigcommerce.com/docs/rest-management/wishlists#get-all-wishlists
148 params = {
149 "limit": limit or DEFAULT_LIMIT,
150 }
151 if filter is not None:
152 params.update(filter)
153 return self._make_request_v3("GET", "wishlists", params=params)
155 def get_wishlists_count(self) -> int:
156 response = self._make_request("GET", urljoin(self.base_url, "v3/wishlists"), params={"limit": 1})
157 return response.get("meta", {}).get("pagination", {}).get("total", 0)
159 def get_segments(
160 self,
161 filter: dict = None,
162 limit: int = None,
163 ):
164 # doc: https://developer.bigcommerce.com/docs/rest-management/customer-segmentation/segments#get-all-segments
165 params = {
166 "limit": limit or DEFAULT_LIMIT,
167 }
168 if filter is not None:
169 params.update(filter)
170 return self._make_request_v3("GET", "segments", params=params)
172 def get_segments_count(self) -> int:
173 response = self._make_request("GET", urljoin(self.base_url, "v3/segments"), params={"limit": 1})
174 return response.get("meta", {}).get("pagination", {}).get("total", 0)
176 def get_brands(
177 self,
178 filter: dict = None,
179 sort_condition: dict = None,
180 limit: int = None,
181 ):
182 # doc: https://developer.bigcommerce.com/docs/rest-catalog/brands#get-all-brands
183 params = {
184 "limit": limit or DEFAULT_LIMIT,
185 }
186 if filter is not None:
187 params.update(filter)
188 if sort_condition is not None:
189 params.update(sort_condition)
190 return self._make_request_v3("GET", "catalog/brands", params=params)
192 def get_brands_count(self) -> int:
193 response = self._make_request("GET", urljoin(self.base_url, "v3/catalog/brands"), params={"limit": 1})
194 return response.get("meta", {}).get("pagination", {}).get("total", 0)
196 def _make_request_v2(self, method: str, url: str, *args, **kwargs) -> list[dict]:
197 # NOTE: v2 limit max is 250
198 url = urljoin(urljoin(self.base_url, "v2/"), url)
199 api_limit = 250
200 params = kwargs.pop("params", {})
201 request_limit = params.get("limit", DEFAULT_LIMIT)
202 params["limit"] = min(api_limit, request_limit)
203 current_page = 1
204 response_len = 1
205 data = []
206 while response_len > 0 and len(data) < request_limit:
207 params["page"] = current_page
208 response = self._make_request(method, url, params=params, *args, **kwargs)
209 if isinstance(response, dict):
210 # for "get_count" requests
211 return response
212 current_page += 1
213 response_len = len(response)
214 data += response
215 return data[:request_limit]
217 def _make_request_v3(self, method: str, url: str, *args, **kwargs) -> list[dict]:
218 url = urljoin(urljoin(self.base_url, "v3/"), url)
219 data = []
220 params = kwargs.pop("params", {})
221 current_page = 1
222 total_pages = 1
223 while current_page <= total_pages:
224 params["page"] = current_page
225 response = self._make_request(method, url, params=params, *args, **kwargs)
226 current_page = response["meta"]["pagination"]["current_page"] + 1
227 total_pages = response["meta"]["pagination"]["total_pages"]
228 data += response["data"]
229 return data
231 def _make_request(self, method: str, url: str, params: dict = None, data: dict = None) -> dict:
232 response = self.session.request(method, url, params=params, json=data)
234 if response.status_code == HTTPStatus.NO_CONTENT:
235 return []
236 if response.status_code != HTTPStatus.OK:
237 raise Exception(f"Request failed with status code {response.status_code}: {response.text}")
239 return response.json()