Coverage for mindsdb / integrations / utilities / install.py: 9%

1import os 

2import sys 

3import subprocess 

4from typing import Text, List 

5 

6 

7def install_dependencies(dependencies: List[Text]) -> dict: 

8 """ 

9 Installs the dependencies for a handler by calling the `pip install` command via subprocess. 

10 

11 Args: 

12 dependencies (List[Text]): List of dependencies for the handler. 

13 

14 Returns: 

15 dict: A dictionary containing the success status and an error message if an error occurs. 

16 """ 

17 outs = b'' 

18 errs = b'' 

19 result = { 

20 'success': False, 

21 'error_message': None 

22 } 

23 code = None 

24 

25 try: 

26 # Split the dependencies by parsing the contents of the requirements.txt file. 

27 split_dependencies = parse_dependencies(dependencies) 

28 except FileNotFoundError as file_not_found_error: 

29 result['error_message'] = f"Error parsing dependencies, file not found: {str(file_not_found_error)}" 

30 return result 

31 except Exception as unknown_error: 

32 result['error_message'] = f"Unknown error parsing dependencies: {str(unknown_error)}" 

33 return result 

34 

35 try: 

36 # Install the dependencies using the `pip install` command. 

37 sp = subprocess.Popen( 

38 [sys.executable, '-m', 'pip', 'install', *split_dependencies], 

39 stdout=subprocess.PIPE, 

40 stderr=subprocess.PIPE 

41 ) 

42 code = sp.wait() 

43 outs, errs = sp.communicate(timeout=1) 

44 except subprocess.TimeoutExpired as timeout_error: 

45 sp.kill() 

46 result['error_message'] = f"Timeout error while installing dependencies: {str(timeout_error)}" 

47 return result 

48 except Exception as unknown_error: 

49 result['error_message'] = f"Unknown error while installing dependencies: {str(unknown_error)}" 

50 return result 

51 

52 # Return the result of the installation if successful, otherwise return an error message. 

53 if code != 0: 

54 output = '' 

55 if isinstance(outs, bytes) and len(outs) > 0: 

56 output = output + 'Output: ' + outs.decode() 

57 if isinstance(errs, bytes) and len(errs) > 0: 

58 if len(output) > 0: 

59 output = output + '\n' 

60 output = output + 'Errors: ' + errs.decode() 

61 result['error_message'] = output 

62 else: 

63 result['success'] = True 

64 

65 return result 

66 

67 

68def parse_dependencies(dependencies: List[Text]) -> List[Text]: 

69 """ 

70 Recursively parses dependencies from a list of dependencies given in a requirements.txt file for a handler. 

71 This function will perform the following: 

72 1. Ignore standalone comments. 

73 2. Remove inline comments. 

74 3. Check if the dependency is a path to a requirements file and recursively parse the dependencies from that file. 

75 

76 Args: 

77 dependencies (List[Text]): List of dependencies for a handler as read from the requirements.txt file. 

78 

79 Returns: 

80 List[Text]: List of parsed dependencies for the handler. 

81 """ 

82 # get the path to this script 

83 script_path = os.path.dirname(os.path.realpath(__file__)) 

84 

85 split_dependencies = [] 

86 for dependency in dependencies: 

87 # ignore standalone comments 

88 if dependency.startswith('#'): 

89 continue 

90 

91 # remove inline comments 

92 if '#' in dependency: 

93 dependency = dependency.split('#')[0].strip() 

94 

95 # check if the dependency is a path to a requirements file 

96 if dependency.startswith('-r'): 

97 # get the path to the requirements file 

98 req_path = dependency.split(' ')[1] 

99 # create the absolute path to the requirements file 

100 abs_req_path = os.path.abspath(os.path.join(script_path, req_path.replace('mindsdb/integrations', '..'))) 

101 # check if the file exists 

102 if os.path.exists(abs_req_path): 

103 inner_dependencies, inner_split_dependencies = [], [] 

104 # read the dependencies from the file 

105 inner_dependencies = read_dependencies(abs_req_path) 

106 # recursively split the dependencies 

107 inner_split_dependencies = parse_dependencies(inner_dependencies) 

108 # add the inner dependencies to the split dependencies 

109 split_dependencies.extend(inner_split_dependencies) 

110 else: 

111 raise FileNotFoundError(f"Requirements file not found: {req_path}") 

112 

113 else: 

114 split_dependencies.append(dependency) 

115 

116 return split_dependencies 

117 

118 

119def read_dependencies(path: Text) -> List[Text]: 

120 """ 

121 Reads the dependencies for a handler from the relevant requirements.txt file and returns them as a list. 

122 

123 Args: 

124 path (Text): Path to the requirements.txt file for the handler. 

125 

126 Returns: 

127 List[Text]: List of dependencies for the handler. 

128 """ 

129 dependencies = [] 

130 # read the dependencies from the file 

131 with open(str(path), 'rt') as f: 

132 dependencies = [x.strip(' \t\n') for x in f.readlines()] 

133 dependencies = [x for x in dependencies if len(x) > 0] 

134 return dependencies