Coverage for pepdbagent/models.py: 98%

1# file with pydantic models 

2import datetime 

3 

4from peprs.const import CONFIG_KEY, SAMPLE_RAW_DICT_KEY, SUBSAMPLE_RAW_DICT_KEY 

5from pydantic import BaseModel, ConfigDict, Field, field_validator 

6 

7from pepdbagent.const import DEFAULT_TAG 

8 

9 

10class ProjectDict(BaseModel): 

11 """ 

12 Project dict (raw) model 

13 """ 

14 

15 config: dict = Field(alias=CONFIG_KEY) 

16 subsample_list: list | None = Field(alias=SUBSAMPLE_RAW_DICT_KEY) 

17 sample_dict: list = Field(alias=SAMPLE_RAW_DICT_KEY) 

18 

19 model_config = ConfigDict(populate_by_name=True, extra="forbid") 

20 

21 

22class AnnotationModel(BaseModel): 

23 """ 

24 Project Annotation model. All meta metadata 

25 """ 

26 

27 namespace: str | None = None 

28 name: str | None = None 

29 tag: str | None = None 

30 is_private: bool | None = None 

31 number_of_samples: int | None = None 

32 description: str | None = None 

33 last_update_date: str | None = None 

34 submission_date: str | None = None 

35 digest: str | None = None 

36 pep_schema: str | None = None 

37 pop: bool | None = False 

38 stars_number: int | None = 0 

39 forked_from: str | None = None 

40 

41 model_config = ConfigDict( 

42 validate_assignment=True, 

43 populate_by_name=True, 

44 ) 

45 

46 @field_validator("is_private") 

47 def is_private_should_be_bool(cls, v) -> bool: 

48 if not isinstance(v, bool): 

49 return False 

50 else: 

51 return v 

52 

53 

54class PaginationResult(BaseModel): 

55 page: int = 0 

56 page_size: int = 10 

57 total: int 

58 

59 

60class AnnotationList(BaseModel): 

61 """ 

62 Annotation return model. 

63 """ 

64 

65 count: int 

66 limit: int 

67 offset: int 

68 results: list[AnnotationModel | None] 

69 

70 

71class Namespace(BaseModel): 

72 """ 

73 Model of single namespace search result 

74 """ 

75 

76 namespace: str 

77 number_of_projects: int 

78 number_of_samples: int 

79 

80 

81class NamespaceList(BaseModel): 

82 """ 

83 Model of combined namespace search results 

84 """ 

85 

86 count: int 

87 limit: int 

88 offset: int 

89 results: list[Namespace] 

90 

91 

92class UpdateItems(BaseModel): 

93 """ 

94 Model used for updating individual items in db 

95 """ 

96 

97 name: str | None = None 

98 description: str | None = None 

99 tag: str | None = None 

100 is_private: bool | None = None 

101 pep_schema: str | None = None 

102 digest: str | None = None 

103 config: dict | None = None 

104 samples: list[dict] | None = None 

105 subsamples: list[list[dict]] | None = None 

106 pop: bool | None = None 

107 schema_id: int | None = None 

108 

109 model_config = ConfigDict( 

110 arbitrary_types_allowed=True, 

111 extra="forbid", 

112 ) 

113 

114 @property 

115 def number_of_samples(self) -> int | None: 

116 if self.samples: 

117 return len(self.samples) 

118 return None 

119 

120 

121class UpdateModel(BaseModel): 

122 """ 

123 !! is Used only by pepdbagent. Don't use it outside 

124 Model used for updating individual items and creating sql string in the code 

125 """ 

126 

127 config: dict | None = None 

128 name: str | None = None 

129 tag: str | None = None 

130 private: bool | None = Field(alias="is_private", default=None) 

131 digest: str | None = None 

132 number_of_samples: int | None = None 

133 pep_schema: str | None = None 

134 description: str | None = "" 

135 # last_update_date: datetime.datetime | None = datetime.datetime.now(datetime.timezone.utc) 

136 pop: bool | None = False 

137 

138 @field_validator("tag", "name") 

139 def value_must_not_be_empty(cls, v) -> str | None: 

140 if "" == v: 

141 return None 

142 return v 

143 

144 @field_validator("tag", "name") 

145 def value_must_be_lowercase(cls, v) -> str | None: 

146 if v: 

147 return v.lower() 

148 return v 

149 

150 @field_validator("tag", "name") 

151 def value_should_not_contain_question(cls, v) -> str: 

152 if "?" in v: 

153 return ValueError("Question mark (?) is prohibited in name and tag.") 

154 return v 

155 

156 model_config = ConfigDict(populate_by_name=True, extra="forbid") 

157 

158 

159class NamespaceInfo(BaseModel): 

160 """ 

161 Model with information about namespace 

162 """ 

163 

164 namespace_name: str 

165 contact_url: str | None = None 

166 number_of_projects: int 

167 number_of_schemas: int 

168 

169 

170class ListOfNamespaceInfo(BaseModel): 

171 """ 

172 Namespace information response model 

173 """ 

174 

175 pagination: PaginationResult 

176 results: list[NamespaceInfo] 

177 

178 

179class ProjectRegistryPath(BaseModel): 

180 """ 

181 Project Namespace 

182 """ 

183 

184 namespace: str 

185 name: str 

186 tag: str = DEFAULT_TAG 

187 

188 

189class ViewAnnotation(BaseModel): 

190 """ 

191 Project views model 

192 """ 

193 

194 name: str 

195 description: str | None = None 

196 number_of_samples: int = 0 

197 

198 

199class ProjectViews(BaseModel): 

200 """ 

201 View annotation model 

202 """ 

203 

204 namespace: str 

205 name: str 

206 tag: str = DEFAULT_TAG 

207 views: list[ViewAnnotation] = [] 

208 

209 

210class CreateViewDictModel(BaseModel): 

211 """ 

212 View creation dict model 

213 """ 

214 

215 project_namespace: str 

216 project_name: str 

217 project_tag: str 

218 sample_list: list[str] 

219 

220 

221class RegistryPath(BaseModel): 

222 namespace: str 

223 name: str 

224 tag: str | None = "default" 

225 

226 

227class NamespaceStats(BaseModel): 

228 """ 

229 Namespace stats model 

230 """ 

231 

232 namespace: str | None = None 

233 projects_updated: dict[str, int] | None = None 

234 projects_created: dict[str, int] | None = None 

235 

236 

237class HistoryChangeModel(BaseModel): 

238 """ 

239 Model for history change 

240 """ 

241 

242 change_id: int 

243 change_date: datetime.datetime 

244 user: str 

245 

246 

247class HistoryAnnotationModel(BaseModel): 

248 """ 

249 History annotation model 

250 """ 

251 

252 namespace: str 

253 name: str 

254 tag: str = DEFAULT_TAG 

255 history: list[HistoryChangeModel] 

256 

257 

258class SchemaVersionAnnotation(BaseModel): 

259 """ 

260 Schema version annotation model 

261 """ 

262 

263 namespace: str 

264 schema_name: str 

265 version: str 

266 contributors: str | None = "" 

267 release_notes: str | None = "" 

268 tags: dict[str, str | None] = {} 

269 release_date: datetime.datetime 

270 last_update_date: datetime.datetime 

271 

272 

273class SchemaRecordAnnotation(BaseModel): 

274 """ 

275 Schema annotation model 

276 """ 

277 

278 namespace: str 

279 schema_name: str 

280 description: str | None = "" 

281 maintainers: str | None = "" 

282 lifecycle_stage: str | None = "" 

283 latest_released_version: str | None = None 

284 private: bool | None = False 

285 last_update_date: datetime.datetime 

286 

287 

288class SchemaSearchResult(BaseModel): 

289 """ 

290 Schema search result model 

291 """ 

292 

293 pagination: PaginationResult 

294 results: list[SchemaRecordAnnotation] 

295 

296 

297class SchemaVersionSearchResult(BaseModel): 

298 """ 

299 Schema version search result model 

300 """ 

301 

302 pagination: PaginationResult 

303 results: list[SchemaVersionAnnotation] 

304 

305 

306class UpdateSchemaRecordFields(BaseModel): 

307 maintainers: str | None = None 

308 lifecycle_stage: str | None = None 

309 private: bool | None = False 

310 name: str | None = None 

311 description: str | None = None 

312 

313 

314class UpdateSchemaVersionFields(BaseModel): 

315 contributors: str | None = None 

316 schema_value: dict | None = None 

317 release_notes: str | None = None 

318 

319 

320class TarNamespaceModel(BaseModel): 

321 """ 

322 Namespace archive model 

323 """ 

324 

325 identifier: int | None = None 

326 namespace: str 

327 file_path: str 

328 creation_date: datetime.datetime | None = None 

329 number_of_projects: int = 0 

330 file_size: int = 0 

331 

332 

333class TarNamespaceModelReturn(BaseModel): 

334 """ 

335 Namespace archive search model 

336 """ 

337 

338 count: int 

339 results: list[TarNamespaceModel]