Coverage for mindsdb / integrations / handlers / bigcommerce_handler / bigcommerce_handler.py: 0%

1from typing import Any 

2 

3from mindsdb.integrations.handlers.bigcommerce_handler.bigcommerce_api_client import BigCommerceAPIClient 

4from mindsdb.integrations.handlers.bigcommerce_handler.bigcommerce_tables import ( 

5 BigCommerceOrdersTable, 

6 BigCommerceProductsTable, 

7 BigCommerceCustomersTable, 

8 BigCommerceCategoriesTable, 

9 BigCommercePickupsTable, 

10 BigCommercePromotionsTable, 

11 BigCommerceWishlistsTable, 

12 BigCommerceSegmentsTable, 

13 BigCommerceBrandsTable, 

14) 

15from mindsdb.integrations.libs.api_handler import MetaAPIHandler 

16from mindsdb.integrations.libs.response import ( 

17 HandlerStatusResponse as StatusResponse, 

18) 

19from mindsdb.utilities import log 

20 

21 

22logger = log.getLogger(__name__) 

23 

24 

25class BigCommerceHandler(MetaAPIHandler): 

26 """This handler handles the connection and execution of SQL statements on BigCommerce.""" 

27 

28 name = "bigcommerce" 

29 

30 def __init__(self, name: str, connection_data: dict, **kwargs: Any) -> None: 

31 """ 

32 Initializes the handler. 

33 

34 Args: 

35 name (str): The name of the handler instance. 

36 connection_data (dict): The connection data required to connect to the BigCommerce API. 

37 kwargs: Arbitrary keyword arguments. 

38 """ 

39 super().__init__(name) 

40 self.connection_data = connection_data 

41 self.kwargs = kwargs 

42 

43 self.connection = None 

44 self.is_connected = False 

45 self.thread_safe = True 

46 

47 self._register_table("orders", BigCommerceOrdersTable(self)) 

48 self._register_table("products", BigCommerceProductsTable(self)) 

49 self._register_table("customers", BigCommerceCustomersTable(self)) 

50 self._register_table("categories", BigCommerceCategoriesTable(self)) 

51 self._register_table("pickups", BigCommercePickupsTable(self)) 

52 self._register_table("promotions", BigCommercePromotionsTable(self)) 

53 self._register_table("wishlists", BigCommerceWishlistsTable(self)) 

54 self._register_table("segments", BigCommerceSegmentsTable(self)) 

55 self._register_table("brands", BigCommerceBrandsTable(self)) 

56 

57 def connect(self) -> BigCommerceAPIClient: 

58 """ 

59 Establishes a connection to the BigCommerce API. 

60 

61 Raises: 

62 ValueError: If the required connection parameters are not provided. 

63 

64 Returns: 

65 BigCommerceAPIClient: A connection object to the BigCommerce API. 

66 """ 

67 if self.is_connected is True: 

68 return self.connection 

69 

70 if not all( 

71 key in self.connection_data and self.connection_data.get(key) for key in ["api_base", "access_token"] 

72 ): 

73 raise ValueError("Required parameters (api_base, access_token) must be provided and should not be empty.") 

74 

75 self.connection = BigCommerceAPIClient( 

76 url=self.connection_data.get("api_base"), 

77 access_token=self.connection_data.get("access_token"), 

78 ) 

79 

80 self.is_connected = True 

81 return self.connection 

82 

83 def check_connection(self) -> StatusResponse: 

84 """ 

85 Checks the status of the connection to the BigCommerce API. 

86 

87 Returns: 

88 StatusResponse: An object containing the success status and an error message if an error occurs. 

89 """ 

90 response = StatusResponse(False) 

91 

92 try: 

93 connection = self.connect() 

94 connection.get_products(limit=1) 

95 response.success = True 

96 except Exception as e: 

97 logger.error(f"Error connecting to BigCommerce API: {e}!") 

98 response.error_message = e 

99 

100 self.is_connected = response.success 

101 

102 return response