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

1import datetime 

2import logging 

3from typing import Literal 

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 Initialize BBObjects. 

26 

27 Args: 

28 config: Config object. 

29 """ 

30 self.config = config 

31 self.bed = BedAgentBedFile(self.config) 

32 self.bedset = BedAgentBedSet(self.config) 

33 

34 def get_thumbnail_uri( 

35 self, 

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

37 record_id: str, 

38 result_id: str, 

39 access_id: str = "http", 

40 ) -> str: 

41 """ 

42 Create URL to access a bed- or bedset-associated thumbnail. 

43 

44 Args: 

45 record_type: Table name ["bed", "bedset"]. 

46 record_id: Record identifier. 

47 result_id: Column name (result name). 

48 access_id: Access id (e.g. http, s3, etc.). 

49 

50 Returns: 

51 String with thumbnail. 

52 """ 

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

54 if result.path_thumbnail: 

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

56 

57 else: 

58 _LOGGER.error( 

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

60 ) 

61 raise MissingThumbnailError( 

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

63 ) 

64 

65 def get_object_uri( 

66 self, 

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

68 record_id: str, 

69 result_id: str, 

70 access_id: str, 

71 ) -> str: 

72 """ 

73 Create URL to access a bed- or bedset-associated file. 

74 

75 Args: 

76 record_type: Table name ["bed", "bedset"]. 

77 record_id: Record identifier. 

78 result_id: Column name (result name). 

79 access_id: Access id (e.g. http, s3, etc.). 

80 

81 Returns: 

82 URI string for the object. 

83 """ 

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

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

86 

87 def _get_result( 

88 self, 

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

90 record_id: str, 

91 result_id: str | list[str], 

92 ) -> FileModel: 

93 """ 

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

95 

96 Args: 

97 record_type: Table name ["bed", "bedset"]. 

98 record_id: Record identifier. 

99 result_id: Column name (result name). e.g. "bigbedfile", "bed_file", "open_chromatin". 

100 

101 Returns: 

102 FileModel instance representing the requested bed/bedset file object. 

103 """ 

104 if record_type == "bed": 

105 try: 

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

107 except KeyError: 

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

109 raise MissingObjectError( 

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

111 ) 

112 elif record_type == "bedset": 

113 try: 

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

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

116 except KeyError: 

117 raise MissingObjectError( 

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

119 ) 

120 

121 else: 

122 raise BedBaseConfError( 

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

124 ) 

125 

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

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

128 return result 

129 

130 def get_drs_metadata( 

131 self, 

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

133 record_id: str, 

134 result_id: str, 

135 base_uri: str, 

136 ) -> DRSModel: 

137 """ 

138 Get DRS metadata for a bed- or bedset-associated file. 

139 

140 Args: 

141 record_type: Bed or bedset. 

142 record_id: Record identifier. 

143 result_id: Name of the result file to get metadata for. 

144 base_uri: Base uri to use for the self_uri field (server hostname of DRS broker). 

145 

146 Returns: 

147 DRS metadata. 

148 """ 

149 

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

151 bed_result = self.bed.get(record_id) 

152 created_time = bed_result.submission_date 

153 modified_time = bed_result.last_update_date 

154 record_metadata = self._get_result( 

155 record_type, record_id, result_id 

156 ) # only get result once 

157 if not record_metadata: 

158 raise MissingObjectError("Record not found") 

159 

160 drs_dict = self.construct_drs_metadata( 

161 base_uri, 

162 object_id, 

163 record_metadata, 

164 created_time, 

165 modified_time, 

166 ) 

167 

168 return drs_dict 

169 

170 def construct_drs_metadata( 

171 self, 

172 base_uri: str, 

173 object_id: str, 

174 record_metadata: FileModel, 

175 created_time: datetime.datetime | None = None, 

176 modified_time: datetime.datetime | None = None, 

177 ) -> DRSModel: 

178 """ 

179 Construct DRS metadata object. 

180 

181 Args: 

182 base_uri: Base uri to use for the self_uri field (server hostname of DRS broker). 

183 object_id: Record identifier. 

184 record_metadata: Metadata of the record. 

185 created_time: Time of creation. 

186 modified_time: Time of last modification. 

187 

188 Returns: 

189 DRS metadata. 

190 """ 

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

192 drs_dict = DRSModel( 

193 id=object_id, 

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

195 size=record_metadata.size or None, 

196 created_time=created_time, 

197 updated_time=modified_time, 

198 checksums=object_id, 

199 access_methods=access_methods, 

200 ) 

201 return drs_dict