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

1from mindsdb.integrations.handlers.instatus_handler.instatus_tables import StatusPages, Components 

2from mindsdb.integrations.libs.api_handler import APIHandler 

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

4from mindsdb.utilities import log 

5from mindsdb_sql_parser import parse_sql 

6import requests 

7import pandas as pd 

8from collections import OrderedDict 

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

10 

11logger = log.getLogger(__name__) 

12 

13 

14class InstatusHandler(APIHandler): 

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

16 """initializer method 

17 

18 Args: 

19 name (str): handler name 

20 """ 

21 super().__init__(name) 

22 self._base_url = "https://api.instatus.com" 

23 self._api_key = None 

24 

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

26 if 'api_key' in args: 

27 self._api_key = args['api_key'] 

28 

29 self.connection = None 

30 self.is_connected = False 

31 

32 _tables = [ 

33 StatusPages, 

34 Components 

35 ] 

36 

37 for Table in _tables: 

38 self._register_table(Table.name, Table(self)) 

39 

40 def check_connection(self) -> StatusResponse: 

41 """checking the connection 

42 

43 Returns: 

44 StatusResponse: whether the connection is still up 

45 """ 

46 response = StatusResponse(False) 

47 

48 try: 

49 self.connect() 

50 response.success = True 

51 except Exception as e: 

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

53 response.error_message = e 

54 

55 self.is_connected = response.success 

56 return response 

57 

58 def connect(self) -> StatusResponse: 

59 # If already connected, return the existing connection 

60 if self.is_connected and self.connection: 

61 return self.connection 

62 

63 if self._api_key: 

64 try: 

65 headers = {"Authorization": f"Bearer {self._api_key}"} 

66 response = requests.get(f"{self._base_url}/v2/pages", headers=headers) 

67 

68 if response.status_code == 200: 

69 self.connection = response 

70 self.is_connected = True 

71 return StatusResponse(True) 

72 else: 

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

74 except requests.RequestException as e: 

75 raise Exception(f"Request to Instatus API failed: {str(e)}") 

76 

77 raise Exception("API key is missing") 

78 

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

80 """Receive and process a raw query. 

81 

82 Parameters 

83 ---------- 

84 query : str 

85 query in a native format 

86 

87 Returns 

88 ------- 

89 StatusResponse 

90 Request status 

91 """ 

92 ast = parse_sql(query) 

93 return self.query(ast) 

94 

95 def call_instatus_api(self, endpoint: str, method: str = 'GET', params: dict = None, json_data: dict = {}) -> pd.DataFrame: 

96 if not params: 

97 params = {} 

98 

99 headers = {"Authorization": f"Bearer {self._api_key}"} 

100 url = f"{self._base_url}{endpoint}" 

101 

102 if method.upper() in ('GET', 'POST', 'PUT', 'DELETE'): 

103 headers['Content-Type'] = 'application/json' 

104 

105 response = requests.request(method, url, headers=headers, params=params, json=json_data) 

106 

107 if response.status_code == 200: 

108 data = response.json() 

109 return pd.DataFrame(data) if isinstance(data, list) else pd.DataFrame([data]) 

110 else: 

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

112 

113 return pd.DataFrame() 

114 

115 

116connection_args = OrderedDict( 

117 api_key={ 

118 "type": ARG_TYPE.PWD, 

119 "description": "Instatus API key to use for authentication.", 

120 "required": True, 

121 "label": "Api key", 

122 }, 

123) 

124 

125connection_args_example = OrderedDict( 

126 api_key="d25509b171ad79395dc2c51b099ee6d0" 

127)