Coverage for pepdbagent/utils.py: 88%

1import datetime 

2import json 

3import uuid 

4from collections.abc import Iterable 

5from hashlib import md5 

6 

7import ubiquerg 

8from peprs.const import SAMPLE_RAW_DICT_KEY 

9 

10from pepdbagent.exceptions import RegistryPathError 

11 

12 

13def is_valid_registry_path(rpath: str) -> bool: 

14 """Verify that a registry path is valid. 

15 

16 Checks that the path contains a forward slash dividing two non-empty strings. 

17 

18 Args: 

19 rpath: Registry path to test. 

20 

21 Returns: 

22 True if the path is a valid registry path. 

23 """ 

24 # check for string 

25 if not isinstance(rpath, str): 

26 return False 

27 return all( 

28 [ 

29 "/" in rpath, 

30 len(rpath.split("/")) == 2, 

31 all([isinstance(s, str) for s in rpath.split("/")]), 

32 ] 

33 ) 

34 

35 

36def all_elements_are_strings(iterable: Iterable) -> bool: 

37 """Check if every element of an iterable is a str. 

38 

39 Args: 

40 iterable: An iterable item. 

41 

42 Returns: 

43 True if every element is a string. 

44 """ 

45 if not isinstance(iterable, Iterable): 

46 return False 

47 return all([isinstance(item, str) for item in iterable]) 

48 

49 

50def create_digest(project_dict: dict) -> str: 

51 """Create an MD5 digest for a PEP project. 

52 

53 Args: 

54 project_dict: Project dictionary. 

55 

56 Returns: 

57 MD5 hex digest of the sample table. 

58 """ 

59 sample_digest = md5( 

60 json.dumps( 

61 project_dict[SAMPLE_RAW_DICT_KEY], 

62 separators=(",", ":"), 

63 ensure_ascii=False, 

64 allow_nan=False, 

65 sort_keys=True, 

66 ).encode("utf-8") 

67 ).hexdigest() 

68 return sample_digest 

69 

70 

71def registry_path_converter(registry_path: str) -> tuple[str, str, str]: 

72 """Convert a registry path to (namespace, name, tag). 

73 

74 Args: 

75 registry_path: Registry path with structure "namespace/name:tag". 

76 

77 Returns: 

78 Tuple of (namespace, name, tag). 

79 

80 Raises: 

81 RegistryPathError: If the path is not a valid registry path. 

82 """ 

83 if is_valid_registry_path(registry_path): 

84 reg = ubiquerg.parse_registry_path(registry_path) 

85 namespace = reg["namespace"] 

86 name = reg["item"] 

87 tag = reg["tag"] 

88 return namespace, name, tag 

89 

90 raise RegistryPathError(f"Error in: '{registry_path}'") 

91 

92 

93def schema_path_converter(schema_path: str) -> tuple[str, str, str]: 

94 """Convert a schema path to (namespace, name, version). 

95 

96 Args: 

97 schema_path: Schema path with structure "namespace/name:version". 

98 

99 Returns: 

100 Tuple of (namespace, name, version). 

101 

102 Raises: 

103 RegistryPathError: If the path is invalid. 

104 """ 

105 if "/" in schema_path: 

106 namespace, name_tag = schema_path.split("/") 

107 if ":" in name_tag: 

108 name, version = name_tag.split(":") 

109 return namespace, name, version 

110 

111 return namespace, name_tag, "latest" 

112 raise RegistryPathError(f"Error in: '{schema_path}'") 

113 

114 

115def tuple_converter(value: tuple | list | str | None) -> tuple: 

116 """Convert a string, list, or tuple to a tuple. 

117 

118 Args: 

119 value: Value to convert. 

120 

121 Returns: 

122 Tuple of strings. 

123 """ 

124 if isinstance(value, str): 

125 value = [value] 

126 if value: 

127 return tuple(value) 

128 return tuple( 

129 " ", 

130 ) 

131 

132 

133def convert_date_string_to_date(date_string: str) -> datetime.datetime: 

134 """Convert a date string to a datetime. 

135 

136 Args: 

137 date_string: Date string in format YYYY/MM/DD, e.g., "2022/02/22". 

138 

139 Returns: 

140 Parsed datetime offset by one day. 

141 """ 

142 return datetime.datetime.strptime(date_string, "%Y/%m/%d") + datetime.timedelta( 

143 days=1 

144 ) 

145 

146 

147def order_samples(results: dict) -> list[dict]: 

148 """Order samples by their parent_guid chain. 

149 

150 # TODO: To make this function more efficient, we should write it in Rust! 

151 

152 Args: 

153 results: Dict of samples keyed by guid, each value a dict with 

154 "sample", "guid", and "parent_guid" keys. 

155 

156 Returns: 

157 Ordered list of samples from root to leaf. 

158 """ 

159 # Find the Root Node 

160 # Create a lookup dictionary for nodes by their GUIDs 

161 guid_lookup = {entry["guid"]: entry for entry in results.values()} 

162 

163 # Create a dictionary to map each GUID to its child GUID 

164 parent_to_child = { 

165 entry["parent_guid"]: entry["guid"] 

166 for entry in results.values() 

167 if entry["parent_guid"] is not None 

168 } 

169 

170 # Find the root node 

171 root = None 

172 for guid, entry in results.items(): 

173 if entry["parent_guid"] is None: 

174 root = entry 

175 break 

176 

177 if root is None: 

178 raise ValueError("No root node found") 

179 

180 ordered_sequence = [] 

181 current = root 

182 

183 while current is not None: 

184 ordered_sequence.append(current) 

185 current_guid = current["guid"] 

186 if current_guid in parent_to_child: 

187 current = guid_lookup[parent_to_child[current_guid]] 

188 else: 

189 current = None 

190 return ordered_sequence 

191 

192 

193def generate_guid() -> str: 

194 return str(uuid.uuid4())