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

1import os 

2from qbosdk import QuickbooksOnlineSDK 

3from mindsdb.integrations.libs.api_handler import APIHandler 

4from mindsdb.integrations.libs.response import ( 

5 HandlerStatusResponse as StatusResponse 

6) 

7from mindsdb.utilities.config import Config 

8from mindsdb.utilities import log 

9from mindsdb_sql_parser import parse_sql 

10from .quickbooks_table import AccountsTable, PurchasesTable, BillPaymentsTable, VendorsTable, BillsTable, EmployeesTable 

11 

12logger = log.getLogger(__name__) 

13 

14 

15class QuickbooksHandler(APIHandler): 

16 """ 

17 A class for handling connections and interactions with Quickbooks API. 

18 """ 

19 

20 def __init__(self, name=None, **kwargs): 

21 super().__init__(name) 

22 

23 args = kwargs.get('connection_data', {}) 

24 

25 self.connection_args = {} 

26 handler_config = Config().get('quickbooks_handler', {}) 

27 for k in ['client_id', 'client_secret', 'refresh_token', 'realm_id', 'environment']: 

28 if k in args: 

29 self.connection_args[k] = args[k] 

30 elif f'QUICKBOOKS_{k.upper()}' in os.environ: 

31 self.connection_args[k] = os.environ[f'QUICKBOOKS_{k.upper()}'] 

32 elif k in handler_config: 

33 self.connection_args[k] = handler_config[k] 

34 

35 self.quickbooks = None 

36 self.is_connected = False 

37 

38 accountso = AccountsTable(self) 

39 self._register_table('accountso', accountso) 

40 purchases = PurchasesTable(self) 

41 self._register_table('purchases', purchases) 

42 bills_payments = BillPaymentsTable(self) 

43 self._register_table('bills_payments', bills_payments) 

44 vendors = VendorsTable(self) 

45 self._register_table('vendors', vendors) 

46 bills = BillsTable(self) 

47 self._register_table('bills', bills) 

48 employees = EmployeesTable(self) 

49 self._register_table('employees', employees) 

50 

51 def connect(self): 

52 if self.is_connected is True: 

53 return self.quickbooks 

54 

55 self.quickbooks = QuickbooksOnlineSDK( 

56 client_id=self.connection_args['client_id'], 

57 client_secret=self.connection_args['client_secret'], 

58 realm_id=self.connection_args['realm_id'], 

59 refresh_token=self.connection_args['refresh_token'], 

60 environment=self.connection_args['environment'] 

61 ) 

62 self.is_connected = True 

63 return self.quickbooks 

64 

65 def check_connection(self) -> StatusResponse: 

66 response = StatusResponse(False) 

67 

68 try: 

69 quickbooks = self.connect() 

70 quickbooks.accounts.get() 

71 logger.info(quickbooks.accounts.get()) 

72 response.success = True 

73 

74 except Exception as e: 

75 response.error_message = f'Error connecting to Quickbooks API: {e}. ' 

76 logger.error(response.error_message) 

77 

78 if response.success is False and self.is_connected is True: 

79 self.is_connected = False 

80 

81 return response 

82 

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

84 """Receive and process a raw query. 

85 Parameters 

86 ---------- 

87 query : str 

88 query in a native format 

89 Returns 

90 ------- 

91 StatusResponse 

92 Request status 

93 """ 

94 ast = parse_sql(query) 

95 return self.query(ast)