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

1from mindsdb_sql_parser import parse_sql 

2 

3from mindsdb.integrations.handlers.luma_handler.luma_tables import ( 

4 LumaEventsTable 

5) 

6from mindsdb.integrations.handlers.luma_handler.luma import LumaClient 

7from mindsdb.integrations.libs.api_handler import APIHandler 

8from mindsdb.integrations.libs.response import ( 

9 HandlerStatusResponse as StatusResponse, 

10) 

11 

12 

13class LumaHandler(APIHandler): 

14 """The Luma handler implementation""" 

15 

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

17 """Initialize the Luma handler. 

18 

19 Parameters 

20 ---------- 

21 name : str 

22 name of a handler instance 

23 """ 

24 super().__init__(name) 

25 

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

27 self.connection_data = connection_data 

28 self.kwargs = kwargs 

29 self.luma_client = None 

30 self.is_connected = False 

31 

32 events_data = LumaEventsTable(self) 

33 self._register_table("events", events_data) 

34 

35 def connect(self) -> StatusResponse: 

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

37 

38 Returns 

39 ------- 

40 StatusResponse 

41 connection object 

42 """ 

43 resp = StatusResponse(False) 

44 try: 

45 self.luma_client = LumaClient(self.connection_data.get("api_key")) 

46 resp.success = True 

47 self.is_connected = True 

48 except Exception as ex: 

49 resp.success = False 

50 resp.error_message = ex 

51 return resp 

52 

53 def check_connection(self) -> StatusResponse: 

54 """Check connection to the handler. 

55 

56 Returns 

57 ------- 

58 StatusResponse 

59 Status confirmation 

60 """ 

61 return self.connect() 

62 

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

64 """Receive and process a raw query. 

65 

66 Parameters 

67 ---------- 

68 query : str 

69 query in a native format 

70 

71 Returns 

72 ------- 

73 StatusResponse 

74 Request status 

75 """ 

76 ast = parse_sql(query) 

77 return self.query(ast)