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

1from mindsdb_sql_parser import parse_sql 

2 

3from mindsdb.integrations.handlers.zendesk_handler.zendesk_tables import ( 

4 ZendeskUsersTable, 

5 ZendeskTicketsTable, 

6 ZendeskTriggersTable, 

7 ZendeskActivitiesTable 

8) 

9from mindsdb.integrations.libs.api_handler import APIHandler 

10from mindsdb.integrations.libs.response import ( 

11 HandlerStatusResponse as StatusResponse, 

12) 

13from mindsdb.utilities import log 

14import zenpy 

15 

16logger = log.getLogger(__name__) 

17 

18 

19class ZendeskHandler(APIHandler): 

20 """The Zendesk handler implementation""" 

21 

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

23 """Initialize the zendesk handler. 

24 

25 Parameters 

26 ---------- 

27 name : str 

28 name of a handler instance 

29 """ 

30 super().__init__(name) 

31 

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

33 self.connection_data = connection_data 

34 self.kwargs = kwargs 

35 self.zen_client = None 

36 self.is_connected = False 

37 

38 self._register_table("users", ZendeskUsersTable(self)) 

39 self._register_table("tickets", ZendeskTicketsTable(self)) 

40 self._register_table("triggers", ZendeskTriggersTable(self)) 

41 self._register_table("activities", ZendeskActivitiesTable(self)) 

42 

43 def connect(self) -> StatusResponse: 

44 """Set up the connection required by the handler. 

45 

46 Returns 

47 ------- 

48 StatusResponse 

49 connection object 

50 """ 

51 resp = StatusResponse(False) 

52 self.zen_client = zenpy.Zenpy(subdomain=self.connection_data["sub_domain"], email=self.connection_data["email"], token=self.connection_data["api_key"]) 

53 try: 

54 self.zen_client.users() 

55 self.is_connected = True 

56 resp.success = True 

57 except Exception as ex: 

58 resp.success = False 

59 resp.error_message = str(ex) 

60 self.is_connected = False 

61 return resp 

62 

63 def check_connection(self) -> StatusResponse: 

64 """Check connection to the handler. 

65 

66 Returns 

67 ------- 

68 StatusResponse 

69 Status confirmation 

70 """ 

71 response = self.connect() 

72 self.is_connected = response.success 

73 return response 

74 

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

76 """Receive and process a raw query. 

77 

78 Parameters 

79 ---------- 

80 query : str 

81 query in a native format 

82 

83 Returns 

84 ------- 

85 StatusResponse 

86 Request status 

87 """ 

88 ast = parse_sql(query) 

89 return self.query(ast)