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

1import gitlab 

2 

3from mindsdb.integrations.handlers.gitlab_handler.gitlab_tables import GitlabIssuesTable, GitlabMergeRequestsTable 

4from mindsdb.integrations.libs.api_handler import APIHandler 

5from mindsdb.integrations.libs.response import ( 

6 HandlerStatusResponse as StatusResponse, 

7) 

8 

9from mindsdb.utilities import log 

10from mindsdb_sql_parser import parse_sql 

11 

12logger = log.getLogger(__name__) 

13 

14 

15class GitlabHandler(APIHandler): 

16 """The GitLab handler implementation""" 

17 

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

19 """constructor 

20 Args: 

21 name (str): the handler name 

22 """ 

23 super().__init__(name) 

24 

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

26 self.connection_data = connection_data 

27 self.repository = connection_data["repository"] 

28 self.kwargs = kwargs 

29 

30 self.connection = None 

31 self.is_connected = False 

32 

33 gitlab_issues_data = GitlabIssuesTable(self) 

34 gitlab_merge_requests_data = GitlabMergeRequestsTable(self) 

35 self._register_table("issues", gitlab_issues_data) 

36 self._register_table("merge_requests", gitlab_merge_requests_data) 

37 

38 def connect(self) -> StatusResponse: 

39 """Set up the connections required by the handler 

40 Returns: 

41 HandlerStatusResponse 

42 """ 

43 

44 connection_kwargs = {} 

45 

46 if self.connection_data.get("url", None): 

47 connection_kwargs["url"] = self.connection_data["url"] 

48 

49 if self.connection_data.get("api_key", None): 

50 connection_kwargs["private_token"] = self.connection_data["api_key"] 

51 

52 self.connection = gitlab.Gitlab(**connection_kwargs) 

53 self.is_connected = True 

54 

55 return self.connection 

56 

57 def check_connection(self) -> StatusResponse: 

58 """Check connection to the handler 

59 Returns: 

60 HandlerStatusResponse 

61 """ 

62 response = StatusResponse(False) 

63 

64 try: 

65 self.connect() 

66 if self.connection_data.get("api_key", None): 

67 logger.info("Authenticated as user") 

68 else: 

69 logger.info("Proceeding without an API key") 

70 

71 response.success = True 

72 except Exception as e: 

73 logger.error(f"Error connecting to GitLab API: {e}!") 

74 response.error_message = e 

75 

76 self.is_connected = response.success 

77 

78 return response 

79 

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

81 """Receive and process raw query. 

82 Args: 

83 query (str): query in a native format 

84 Returns: 

85 HandlerResponse 

86 """ 

87 ast = parse_sql(query) 

88 return self.query(ast)