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

1import stripe 

2from mindsdb.integrations.handlers.stripe_handler.stripe_tables import CustomersTable, ProductsTable, PaymentIntentsTable, RefundsTable, PayoutsTable 

3from mindsdb.integrations.libs.api_handler import APIHandler 

4from mindsdb.integrations.libs.response import ( 

5 HandlerStatusResponse as StatusResponse, 

6) 

7 

8from mindsdb.utilities import log 

9from mindsdb_sql_parser import parse_sql 

10 

11logger = log.getLogger(__name__) 

12 

13 

14class StripeHandler(APIHandler): 

15 """ 

16 The Stripe handler implementation. 

17 """ 

18 

19 name = 'stripe' 

20 

21 def __init__(self, name: str, **kwargs): 

22 """ 

23 Initialize the handler. 

24 Args: 

25 name (str): name of particular handler instance 

26 **kwargs: arbitrary keyword arguments. 

27 """ 

28 super().__init__(name) 

29 

30 connection_data = kwargs.get("connection_data", {}) 

31 self.connection_data = connection_data 

32 self.kwargs = kwargs 

33 

34 self.connection = None 

35 self.is_connected = False 

36 

37 customers_data = CustomersTable(self) 

38 self._register_table("customers", customers_data) 

39 

40 products_data = ProductsTable(self) 

41 self._register_table("products", products_data) 

42 

43 payment_intents_data = PaymentIntentsTable(self) 

44 self._register_table("payment_intents", payment_intents_data) 

45 

46 payouts_data = PayoutsTable(self) 

47 self._register_table("payouts", payouts_data) 

48 

49 refunds_data = RefundsTable(self) 

50 self._register_table("refunds", refunds_data) 

51 

52 def connect(self): 

53 """ 

54 Set up the connection required by the handler. 

55 Returns 

56 ------- 

57 StatusResponse 

58 connection object 

59 """ 

60 if self.is_connected is True: 

61 return self.connection 

62 

63 stripe.api_key = self.connection_data['api_key'] 

64 

65 self.connection = stripe 

66 self.is_connected = True 

67 

68 return self.connection 

69 

70 def check_connection(self) -> StatusResponse: 

71 """ 

72 Check connection to the handler. 

73 Returns: 

74 HandlerStatusResponse 

75 """ 

76 

77 response = StatusResponse(False) 

78 

79 try: 

80 stripe = self.connect() 

81 stripe.Account.retrieve() 

82 response.success = True 

83 except Exception as e: 

84 logger.error('Error connecting to Stripe!') 

85 response.error_message = str(e) 

86 

87 self.is_connected = response.success 

88 

89 return response 

90 

91 def native_query(self, query: str) -> StatusResponse: 

92 """Receive and process a raw query. 

93 Parameters 

94 ---------- 

95 query : str 

96 query in a native format 

97 Returns 

98 ------- 

99 StatusResponse 

100 Request status 

101 """ 

102 ast = parse_sql(query) 

103 return self.query(ast)