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

1import requests 

2from typing import Any, Dict, List, Optional 

3 

4from mindsdb_sql_parser import parse_sql 

5 

6from mindsdb.integrations.handlers.gong_handler.gong_tables import ( 

7 GongCallsTable, 

8 GongUsersTable, 

9 GongAnalyticsTable, 

10 GongTranscriptsTable, 

11) 

12from mindsdb.integrations.handlers.gong_handler.constants import ( 

13 get_gong_api_info, 

14 GONG_TABLES_METADATA, 

15 GONG_PRIMARY_KEYS, 

16 GONG_FOREIGN_KEYS, 

17) 

18from mindsdb.integrations.libs.api_handler import MetaAPIHandler 

19from mindsdb.integrations.libs.response import ( 

20 HandlerResponse as Response, 

21 HandlerStatusResponse as StatusResponse, 

22 RESPONSE_TYPE, 

23) 

24from mindsdb.utilities import log 

25 

26 

27logger = log.getLogger(__name__) 

28 

29 

30class GongHandler(MetaAPIHandler): 

31 """ 

32 This handler handles the connection and execution of SQL statements on Gong. 

33 """ 

34 

35 name = "gong" 

36 

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

38 """ 

39 Initializes the handler. 

40 

41 Args: 

42 name (Text): The name of the handler instance. 

43 connection_data (Dict): The connection data required to connect to the Gong API. 

44 kwargs: Arbitrary keyword arguments. 

45 """ 

46 super().__init__(name) 

47 self.connection_data = connection_data 

48 self.kwargs = kwargs 

49 

50 self.connection = None 

51 self.is_connected = False 

52 self.base_url = connection_data.get("base_url", "https://api.gong.io") 

53 self.timeout = connection_data.get("timeout", 30) # Default 30 second timeout 

54 

55 # Support both bearer token and access key + secret key 

56 self.bearer_token = connection_data.get("api_key") 

57 self.access_key = connection_data.get("access_key") 

58 self.secret_key = connection_data.get("secret_key") 

59 

60 # Register core tables 

61 self._register_table("calls", GongCallsTable(self)) 

62 self._register_table("users", GongUsersTable(self)) 

63 self._register_table("analytics", GongAnalyticsTable(self)) 

64 self._register_table("transcripts", GongTranscriptsTable(self)) 

65 

66 def connect(self) -> requests.Session: 

67 """ 

68 Establishes a connection to the Gong API. 

69 

70 Raises: 

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

72 Exception: If a connection error occurs. 

73 

74 Returns: 

75 requests.Session: A session object for making API requests. 

76 """ 

77 if self.is_connected is True: 

78 return self.connection 

79 

80 if self.access_key and self.secret_key: 

81 auth_method = "basic" 

82 elif self.bearer_token: 

83 auth_method = "bearer" 

84 else: 

85 raise ValueError("Either bearer_token or (access_key + secret_key) is required to connect to Gong API.") 

86 

87 try: 

88 self.connection = requests.Session() 

89 

90 if auth_method == "basic": 

91 # Basic authentication with access key + secret key 

92 self.connection.auth = (self.access_key, self.secret_key) 

93 self.connection.headers.update({"Content-Type": "application/json", "Accept": "application/json"}) 

94 else: 

95 # Bearer token authentication 

96 self.connection.headers.update( 

97 { 

98 "Authorization": f"Bearer {self.bearer_token}", 

99 "Content-Type": "application/json", 

100 "Accept": "application/json", 

101 } 

102 ) 

103 

104 test_response = self.connection.get(f"{self.base_url}/v2/users", timeout=self.timeout) 

105 test_response.raise_for_status() 

106 

107 self.is_connected = True 

108 return self.connection 

109 

110 except Exception as e: 

111 self.is_connected = False 

112 logger.error(f"Error connecting to Gong API: {e}") 

113 raise 

114 

115 def check_connection(self) -> StatusResponse: 

116 """ 

117 Checks the status of the connection to the Gong API. 

118 

119 Returns: 

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

121 """ 

122 response = StatusResponse(False) 

123 

124 try: 

125 self.connect() 

126 # Test the connection by making a simple API call 

127 test_response = self.connection.get(f"{self.base_url}/v2/users") 

128 test_response.raise_for_status() 

129 response.success = True 

130 except Exception as e: 

131 logger.error(f"Connection check to Gong failed: {e}") 

132 response.error_message = str(e) 

133 

134 self.is_connected = response.success 

135 return response 

136 

137 def native_query(self, query: str) -> Response: 

138 """ 

139 Executes a native query on Gong and returns the result. 

140 

141 Args: 

142 query (Text): The SQL query to be executed. 

143 

144 Returns: 

145 Response: A response object containing the result of the query or an error message. 

146 """ 

147 try: 

148 ast = parse_sql(query) 

149 return self.query(ast) 

150 except Exception as e: 

151 logger.error(f"Error running query: {query} on Gong: {e}") 

152 return Response(RESPONSE_TYPE.ERROR, error_code=0, error_message=str(e)) 

153 

154 def call_gong_api(self, endpoint: str, method: str = "GET", params: Dict = None, json: Dict = None) -> Dict: 

155 """ 

156 Makes a call to the Gong API. 

157 

158 Args: 

159 endpoint (str): The API endpoint to call. 

160 method (str): HTTP method (GET or POST). 

161 params (Dict): Query parameters for the API call (for GET requests). 

162 json (Dict): JSON payload for POST requests. 

163 

164 Returns: 

165 Dict: The API response. 

166 """ 

167 if not self.is_connected: 

168 self.connect() 

169 

170 url = f"{self.base_url}{endpoint}" 

171 

172 if method.upper() == "POST": 

173 response = self.connection.post(url, json=json, timeout=self.timeout) 

174 else: 

175 response = self.connection.get(url, params=params, timeout=self.timeout) 

176 

177 response.raise_for_status() 

178 return response.json() 

179 

180 def meta_get_handler_info(self, **kwargs) -> str: 

181 """ 

182 Retrieves information about the Gong API handler design and implementation. 

183 

184 Returns: 

185 str: A string containing information about the handler's design and implementation. 

186 """ 

187 return get_gong_api_info(self.name) 

188 

189 def meta_get_tables(self, table_names: Optional[List[str]] = None, **kwargs) -> Response: 

190 """ 

191 Retrieves metadata for the specified tables (or all tables if no list is provided). 

192 

193 Note: Gong API doesn't provide a metadata/schema discovery endpoint, so we use 

194 the handler's registered tables combined with static metadata from constants. 

195 

196 Args: 

197 table_names (List): A list of table names for which to retrieve metadata. 

198 

199 Returns: 

200 Response: A response object containing the table metadata. 

201 """ 

202 import pandas as pd 

203 

204 metadata_list = [] 

205 

206 # Get metadata for requested tables (or all if none specified) 

207 # Use registered tables to ensure we only return tables that are actually available 

208 for table_name in self._tables.keys(): 

209 if (table_names is None or table_name in table_names) and table_name in GONG_TABLES_METADATA: 

210 metadata = GONG_TABLES_METADATA[table_name] 

211 metadata_list.append( 

212 { 

213 "table_name": metadata["name"], 

214 "table_type": metadata["type"], 

215 "description": metadata["description"], 

216 "api_endpoint": metadata["api_endpoint"], 

217 "supports_pagination": metadata["supports_pagination"], 

218 "notes": metadata.get("notes", ""), 

219 } 

220 ) 

221 

222 df = pd.DataFrame(metadata_list) 

223 return Response(RESPONSE_TYPE.TABLE, df) 

224 

225 def meta_get_columns(self, table_names: Optional[List[str]] = None, **kwargs) -> Response: 

226 """ 

227 Retrieves column metadata for the specified tables (or all tables if no list is provided). 

228 

229 Note: Column schemas are derived from the table's get_columns() method when available, 

230 falling back to static metadata from constants. 

231 

232 Args: 

233 table_names (List): A list of table names for which to retrieve column metadata. 

234 

235 Returns: 

236 Response: A response object containing the column metadata. 

237 """ 

238 import pandas as pd 

239 

240 column_metadata_list = [] 

241 

242 # Get column metadata for requested tables (or all if none specified) 

243 for table_name in self._tables.keys(): 

244 if (table_names is None or table_name in table_names) and table_name in GONG_TABLES_METADATA: 

245 metadata = GONG_TABLES_METADATA[table_name] 

246 

247 # Try to get live columns from the table class 

248 table_instance = self._tables[table_name] 

249 if hasattr(table_instance, "get_columns"): 

250 live_columns = table_instance.get_columns() 

251 

252 # Match live columns with metadata 

253 for column_name in live_columns: 

254 # Find column metadata 

255 column_meta = next( 

256 (col for col in metadata["columns"] if col["name"] == column_name), 

257 {"type": "str", "description": f"Column {column_name}"}, 

258 ) 

259 

260 column_metadata_list.append( 

261 { 

262 "table_name": table_name, 

263 "column_name": column_name, 

264 "data_type": column_meta.get("type", "str"), 

265 "description": column_meta.get("description", ""), 

266 "is_filterable": column_name in metadata.get("filterable_columns", []), 

267 } 

268 ) 

269 else: 

270 # Fallback to static metadata 

271 for column in metadata["columns"]: 

272 column_metadata_list.append( 

273 { 

274 "table_name": table_name, 

275 "column_name": column["name"], 

276 "data_type": column["type"], 

277 "description": column["description"], 

278 "is_filterable": column["name"] in metadata.get("filterable_columns", []), 

279 } 

280 ) 

281 

282 df = pd.DataFrame(column_metadata_list) 

283 return Response(RESPONSE_TYPE.TABLE, df) 

284 

285 def meta_get_primary_keys(self, table_names: Optional[List[str]] = None, **kwargs) -> Response: 

286 """ 

287 Retrieves primary key metadata for the specified tables (or all tables if no list is provided). 

288 

289 Args: 

290 table_names (List): A list of table names for which to retrieve primary key metadata. 

291 

292 Returns: 

293 Response: A response object containing the primary key metadata. 

294 """ 

295 import pandas as pd 

296 

297 pk_list = [] 

298 

299 # Get primary key metadata for requested tables (or all if none specified) 

300 for table_name in self._tables.keys(): 

301 if (table_names is None or table_name in table_names) and table_name in GONG_PRIMARY_KEYS: 

302 pk_info = GONG_PRIMARY_KEYS[table_name] 

303 pk_list.append( 

304 { 

305 "TABLE_NAME": table_name, 

306 "COLUMN_NAME": pk_info["column_name"], 

307 "CONSTRAINT_NAME": pk_info["constraint_name"], 

308 } 

309 ) 

310 

311 df = pd.DataFrame(pk_list) 

312 return Response(RESPONSE_TYPE.TABLE, df) 

313 

314 def meta_get_foreign_keys(self, table_names: Optional[List[str]] = None, **kwargs) -> Response: 

315 """ 

316 Retrieves foreign key metadata for the specified tables (or all tables if no list is provided). 

317 

318 Args: 

319 table_names (List): A list of table names for which to retrieve foreign key metadata. 

320 

321 Returns: 

322 Response: A response object containing the foreign key metadata. 

323 """ 

324 import pandas as pd 

325 

326 fk_list = [] 

327 

328 # Get foreign key metadata for requested tables (or all if none specified) 

329 for table_name in self._tables.keys(): 

330 if (table_names is None or table_name in table_names) and table_name in GONG_FOREIGN_KEYS: 

331 for fk_info in GONG_FOREIGN_KEYS[table_name]: 

332 fk_list.append( 

333 { 

334 "TABLE_NAME": table_name, 

335 "COLUMN_NAME": fk_info["column_name"], 

336 "FOREIGN_TABLE_NAME": fk_info["foreign_table_name"], 

337 "FOREIGN_COLUMN_NAME": fk_info["foreign_column_name"], 

338 "CONSTRAINT_NAME": fk_info["constraint_name"], 

339 } 

340 ) 

341 

342 df = pd.DataFrame(fk_list) 

343 return Response(RESPONSE_TYPE.TABLE, df)