Coverage for bbconf/modules/objects.py: 32%

1import datetime 

2import logging 

3from typing import List, Literal, Union 

4 

5from bbconf.config_parser.bedbaseconfig import BedBaseConfig 

6from bbconf.const import PKG_NAME 

7from bbconf.exceptions import ( 

8 BedBaseConfError, 

9 MissingObjectError, 

10 MissingThumbnailError, 

11) 

12from bbconf.models.bed_models import FileModel 

13from bbconf.models.drs_models import DRSModel 

14from bbconf.modules.bedfiles import BedAgentBedFile 

15from bbconf.modules.bedsets import BedAgentBedSet 

16 

17_LOGGER = logging.getLogger(PKG_NAME) 

18 

19 

20class BBObjects: 

21 """ """ 

22 

23 def __init__(self, config: BedBaseConfig): 

24 """ 

25 :param config: config object 

26 """ 

27 self.config = config 

28 self.bed = BedAgentBedFile(self.config) 

29 self.bedset = BedAgentBedSet(self.config) 

30 

31 def get_thumbnail_uri( 

32 self, 

33 record_type: Literal["bed", "bedset"], 

34 record_id: str, 

35 result_id: str, 

36 access_id: str = "http", 

37 ) -> str: 

38 """ 

39 Create URL to access a bed- or bedset-associated thumbnail 

40 

41 :param record_type: table_name ["bed", "bedset"] 

42 :param record_id: record identifier 

43 :param result_id: column name (result name) 

44 :param access_id: access id (e.g. http, s3, etc.) 

45 :return: string with thumbnail 

46 """ 

47 result = self._get_result(record_type, record_id, result_id) 

48 if result.path_thumbnail: 

49 return self.config.get_prefixed_uri(result.path_thumbnail, access_id) 

50 

51 else: 

52 _LOGGER.error( 

53 f"Thumbnail for {record_type} {record_id} {result_id} is not defined." 

54 ) 

55 raise MissingThumbnailError( 

56 f"Thumbnail for {record_type} {record_id} {result_id} is not defined." 

57 ) 

58 

59 def get_object_uri( 

60 self, 

61 record_type: Literal["bed", "bedset"], 

62 record_id: str, 

63 result_id: str, 

64 access_id: str, 

65 ) -> str: 

66 """ 

67 Create URL to access a bed- or bedset-associated file 

68 

69 :param record_type: table_name ["bed", "bedset"] 

70 :param record_id: record identifier 

71 :param result_id: column name (result name) 

72 :param access_id: access id (e.g. http, s3, etc.) 

73 :return: 

74 """ 

75 result = self._get_result(record_type, record_id, result_id) 

76 return self.config.get_prefixed_uri(result.path, access_id) 

77 

78 def _get_result( 

79 self, 

80 record_type: Literal["bed", "bedset"], 

81 record_id: str, 

82 result_id: Union[str, List[str]], 

83 ) -> FileModel: 

84 """ 

85 Generic getter that can return a result from either bed or bedset 

86 

87 :param record_type: table_name ["bed", "bedset"] 

88 :param record_id: record identifier 

89 :param result_id: column name (result name). e.g. "bigbedfile", "bed_file", "open_chromatin" 

90 :return: pipestat result 

91 """ 

92 if record_type == "bed": 

93 try: 

94 result = self.bed.get_objects(identifier=record_id)[result_id] 

95 except KeyError: 

96 _LOGGER.error(f"Result {result_id} is not defined for bed {record_id}") 

97 raise MissingObjectError( 

98 f"Result {result_id} is not defined for bed {record_id}" 

99 ) 

100 elif record_type == "bedset": 

101 try: 

102 result = self.bedset.get_objects(identifier=record_id)[result_id] 

103 _LOGGER.error(f"Result {result_id} is not defined for bed {record_id}") 

104 except KeyError: 

105 raise MissingObjectError( 

106 f"Result {result_id} is not defined for bed {record_id}" 

107 ) 

108 

109 else: 

110 raise BedBaseConfError( 

111 f"Record type {record_type} is not supported. Only bed and bedset are supported." 

112 ) 

113 

114 _LOGGER.info(f"Getting uri for {record_type} {record_id} {result_id}") 

115 _LOGGER.debug(f"Result: {result}") 

116 return result 

117 

118 def get_drs_metadata( 

119 self, 

120 record_type: Literal["bed", "bedset"], 

121 record_id: str, 

122 result_id: str, 

123 base_uri: str, 

124 ) -> DRSModel: 

125 """ 

126 Get DRS metadata for a bed- or bedset-associated file 

127 

128 :param record_type: bed or bedset 

129 :param record_id: record identifier 

130 :param result_id: name of the result file to get metadata for 

131 :param base_uri: base uri to use for the self_uri field (server hostname of DRS broker) 

132 :return: DRS metadata 

133 """ 

134 

135 object_id = f"{record_type}.{record_id}.{result_id}" 

136 bed_result = self.bed.get(record_id) 

137 created_time = bed_result.submission_date 

138 modified_time = bed_result.last_update_date 

139 record_metadata = self._get_result( 

140 record_type, record_id, result_id 

141 ) # only get result once 

142 if not record_metadata: 

143 raise MissingObjectError("Record not found") 

144 

145 drs_dict = self.construct_drs_metadata( 

146 base_uri, 

147 object_id, 

148 record_metadata, 

149 created_time, 

150 modified_time, 

151 ) 

152 

153 return drs_dict 

154 

155 def construct_drs_metadata( 

156 self, 

157 base_uri: str, 

158 object_id: str, 

159 record_metadata: FileModel, 

160 created_time: datetime.datetime = None, 

161 modified_time: datetime.datetime = None, 

162 ): 

163 """ 

164 Construct DRS metadata object 

165 

166 :param base_uri: base uri to use for the self_uri field (server hostname of DRS broker) 

167 :param object_id: record identifier 

168 :param record_metadata: metadata of the record 

169 :param created_time: time of creation 

170 :param modified_time: time of last modification 

171 :return: DRS metadata 

172 """ 

173 access_methods = self.config.construct_access_method_list(record_metadata.path) 

174 drs_dict = DRSModel( 

175 id=object_id, 

176 self_uri=f"drs://{base_uri}/{object_id}", 

177 size=record_metadata.size or None, 

178 created_time=created_time, 

179 updated_time=modified_time, 

180 checksums=object_id, 

181 access_methods=access_methods, 

182 ) 

183 return drs_dict