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

1from mindsdb.integrations.handlers.intercom_handler.intercom_tables import Articles 

2from mindsdb.integrations.libs.api_handler import APIHandler 

3from mindsdb.integrations.libs.response import HandlerStatusResponse as StatusResponse 

4from mindsdb_sql_parser import parse_sql 

5import requests 

6import pandas as pd 

7from collections import OrderedDict 

8from mindsdb.integrations.libs.const import HANDLER_CONNECTION_ARG_TYPE as ARG_TYPE 

9import json 

10from mindsdb.utilities import log 

11 

12logger = log.getLogger(__name__) 

13 

14 

15class IntercomHandler(APIHandler): 

16 def __init__(self, name: str, **kwargs) -> None: 

17 """initializer method 

18 

19 Args: 

20 name (str): handler name 

21 """ 

22 super().__init__(name) 

23 

24 self.connection = None 

25 self.is_connected = False 

26 self._baseUrl = 'https://api.intercom.io' 

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

28 if 'access_token' in args: 

29 access_token = args['access_token'] 

30 self._headers = { 

31 "Accept": "application/json", 

32 "Authorization": f"Bearer {access_token}" 

33 } 

34 self._register_table(Articles.name, Articles(self)) 

35 

36 def check_connection(self) -> StatusResponse: 

37 """checking the connection 

38 

39 Returns: 

40 StatusResponse: whether the connection is still up 

41 """ 

42 response = StatusResponse(False) 

43 

44 try: 

45 self.connect() 

46 response.success = True 

47 except Exception as e: 

48 logger.error(f'Error connecting to Intercom API: {e}!') 

49 response.error_message = e 

50 

51 self.is_connected = response.success 

52 return response 

53 

54 def connect(self) -> StatusResponse: 

55 """making the connectino object 

56 """ 

57 if self.is_connected and self.connection: 

58 return self.connection 

59 

60 if self._headers: 

61 try: 

62 response = requests.get( 

63 url=self._baseUrl, 

64 headers=self._headers 

65 ) 

66 if response.status_code == 200: 

67 self.connection = response 

68 self.is_connected = True 

69 return StatusResponse(True) 

70 else: 

71 raise Exception(f"Error connecting to Intercom API: {response.status_code} - {response.text}") 

72 except requests.RequestException as e: 

73 raise Exception(f"Request to Intercom API failed: {str(e)}") 

74 

75 raise Exception("Access token is missing") 

76 

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

78 """Receive and process a raw query. 

79 

80 Parameters 

81 ---------- 

82 query : str 

83 query in a native format 

84 

85 Returns 

86 ------- 

87 StatusResponse 

88 Request status 

89 """ 

90 ast = parse_sql(query) 

91 return self.query(ast) 

92 

93 def call_intercom_api(self, endpoint: str, method: str = 'GET', params: dict = {}, data=None) -> pd.DataFrame: 

94 url = f"{self._baseUrl}{endpoint}" 

95 json_data = json.loads(data) if data else None 

96 

97 response = requests.request(method.upper(), url, headers=self._headers, params=params, json=json_data) 

98 

99 if response.status_code == 200: 

100 data = response.json() 

101 return pd.DataFrame([data]) 

102 else: 

103 raise requests.Response.raise_for_status(response) 

104 

105 

106connection_args = OrderedDict( 

107 access_token={ 

108 "type": ARG_TYPE.PWD, 

109 "description": "Intercom access token to use for authentication.", 

110 "required": True, 

111 "label": "Access token", 

112 }, 

113) 

114 

115connection_args_example = OrderedDict( 

116 api_key="d25509b171ad79395dc2c51b099ee6d0" 

117)