Coverage for pepdbagent/modules/namespace.py: 91%

1import logging 

2from collections import Counter 

3from datetime import datetime, timedelta 

4 

5from sqlalchemy import delete, distinct, func, or_, select 

6from sqlalchemy.orm import Session 

7from sqlalchemy.sql.selectable import Select 

8 

9from pepdbagent.const import DEFAULT_LIMIT, DEFAULT_LIMIT_INFO, DEFAULT_OFFSET, PKG_NAME 

10from pepdbagent.db_utils import BaseEngine, Projects, TarNamespace, User 

11from pepdbagent.exceptions import NamespaceNotFoundError 

12from pepdbagent.models import ( 

13 ListOfNamespaceInfo, 

14 Namespace, 

15 NamespaceInfo, 

16 NamespaceList, 

17 NamespaceStats, 

18 PaginationResult, 

19 TarNamespaceModel, 

20 TarNamespaceModelReturn, 

21) 

22from pepdbagent.utils import tuple_converter 

23 

24_LOGGER = logging.getLogger(PKG_NAME) 

25 

26 

27class PEPDatabaseNamespace: 

28 """ 

29 Class that represents project Namespaces in Database. 

30 

31 While using this class, user can retrieve all necessary metadata about PEPs 

32 """ 

33 

34 def __init__(self, pep_db_engine: BaseEngine): 

35 """ 

36 Args: 

37 pep_db_engine: PEPDatabaseAgent engine object. 

38 """ 

39 self._sa_engine = pep_db_engine.engine 

40 self._pep_db_engine = pep_db_engine 

41 

42 def get( 

43 self, 

44 query: str = "", 

45 admin: list[str] | str | None = None, 

46 limit: int = DEFAULT_LIMIT, 

47 offset: int = DEFAULT_OFFSET, 

48 ) -> NamespaceList: 

49 """Search available namespaces in the database. 

50 

51 Args: 

52 query: Search string. 

53 admin: Namespaces where the user has admin rights. 

54 limit: Maximum number of results to return. 

55 offset: Number of results to skip. 

56 

57 Returns: 

58 NamespaceList with count, limit, offset, and results. 

59 """ 

60 _LOGGER.info( 

61 f"Getting namespaces annotation with provided info: (query: {query})" 

62 ) 

63 admin_tuple = tuple_converter(admin) 

64 return NamespaceList( 

65 count=self._count_namespace(search_str=query, admin_nsp=admin_tuple), 

66 limit=limit, 

67 offset=offset, 

68 results=self._get_namespace( 

69 search_str=query, 

70 admin_nsp=admin_tuple, 

71 limit=limit, 

72 offset=offset, 

73 ), 

74 ) 

75 

76 def _get_namespace( 

77 self, 

78 search_str: str, 

79 admin_nsp: tuple = None, 

80 limit: int = DEFAULT_LIMIT, 

81 offset: int = DEFAULT_OFFSET, 

82 ) -> list[Namespace]: 

83 """Search for namespaces matching a search string. 

84 

85 Args: 

86 search_str: Keywords to search in namespace names. 

87 admin_nsp: Namespaces accessible even when private. 

88 limit: Maximum number of results. 

89 offset: Number of results to skip. 

90 

91 Returns: 

92 List of Namespace objects. 

93 """ 

94 statement = ( 

95 select( 

96 Projects.namespace, 

97 func.count(Projects.name).label("number_of_projects"), 

98 func.sum(Projects.number_of_samples).label("number_of_samples"), 

99 ) 

100 .group_by(Projects.namespace) 

101 .select_from(Projects) 

102 ) 

103 

104 statement = self._add_condition( 

105 statement=statement, 

106 search_str=search_str, 

107 admin_list=admin_nsp, 

108 ) 

109 

110 statement = statement.limit(limit).offset(offset) 

111 

112 with Session(self._sa_engine) as session: 

113 query_results = session.execute(statement).all() 

114 

115 results_list = [] 

116 for res in query_results: 

117 results_list.append( 

118 Namespace( 

119 namespace=res.namespace, 

120 number_of_projects=res.number_of_projects, 

121 number_of_samples=res.number_of_samples, 

122 ) 

123 ) 

124 return results_list 

125 

126 def _count_namespace( 

127 self, search_str: str = None, admin_nsp: tuple = tuple() 

128 ) -> int: 

129 """Count namespaces matching a search string. 

130 

131 Args: 

132 search_str: Keywords to search in namespace names. 

133 admin_nsp: Namespaces accessible even when private. 

134 

135 Returns: 

136 Number of matching namespaces. 

137 """ 

138 statement = select( 

139 func.count(distinct(Projects.namespace)).label("number_of_namespaces") 

140 ).select_from(Projects) 

141 statement = self._add_condition( 

142 statement=statement, 

143 search_str=search_str, 

144 admin_list=admin_nsp, 

145 ) 

146 with Session(self._sa_engine) as session: 

147 query_results = session.execute(statement).one() 

148 

149 return query_results.number_of_namespaces 

150 

151 @staticmethod 

152 def _add_condition( 

153 statement: Select, 

154 search_str: str = None, 

155 admin_list: tuple[str, ...] | list[str] | str | None = None, 

156 ) -> Select: 

157 """Add a WHERE clause to a namespace search statement. 

158 

159 Args: 

160 statement: SQLAlchemy SELECT statement to augment. 

161 search_str: String to search in namespace names. 

162 admin_list: Namespaces with admin/private access. 

163 

164 Returns: 

165 Statement with WHERE clause applied. 

166 """ 

167 if search_str: 

168 sql_search_str = f"%{search_str}%" 

169 statement = statement.where( 

170 or_( 

171 Projects.namespace.ilike(sql_search_str), 

172 ) 

173 ) 

174 statement = statement.where( 

175 or_(Projects.private.is_(False), Projects.namespace.in_(admin_list)) 

176 ) 

177 return statement 

178 

179 def info( 

180 self, 

181 page: int = 0, 

182 page_size: int = DEFAULT_LIMIT_INFO, 

183 order_by: str = "number_of_projects", 

184 ) -> ListOfNamespaceInfo: 

185 """Get a paginated list of top namespaces. 

186 

187 Warning: counts all projects including private ones (by design, for efficiency). 

188 

189 Args: 

190 page: Page number (zero-based). 

191 page_size: Number of namespaces per page. 

192 order_by: Sort field — "number_of_projects" or "number_of_schemas". 

193 

194 Returns: 

195 ListOfNamespaceInfo with pagination metadata and results. 

196 """ 

197 

198 statement = select(User) 

199 

200 if order_by == "number_of_projects": 

201 statement = statement.order_by(User.number_of_projects.desc()) 

202 elif order_by == "number_of_schemas": 

203 statement = statement.order_by(User.number_of_schemas.desc()) 

204 

205 with Session(self._sa_engine) as session: 

206 results = session.scalars( 

207 statement.limit(page_size).offset(page_size * page) 

208 ) 

209 total_number_of_namespaces = session.execute( 

210 select(func.count(User.id)) 

211 ).one()[0] 

212 

213 list_of_results = [] 

214 for result in results: 

215 list_of_results.append( 

216 NamespaceInfo( 

217 namespace_name=result.namespace, 

218 contact_url=f"https://github.com/{result.namespace}", 

219 number_of_projects=result.number_of_projects, 

220 number_of_schemas=result.number_of_schemas, 

221 ) 

222 ) 

223 return ListOfNamespaceInfo( 

224 pagination=PaginationResult( 

225 page=page, 

226 page_size=page_size, 

227 total=total_number_of_namespaces, 

228 ), 

229 results=list_of_results, 

230 ) 

231 

232 def stats(self, namespace: str = None, monthly: bool = False) -> NamespaceStats: 

233 """Get submission/update statistics for a namespace or the whole database. 

234 

235 Args: 

236 namespace: Namespace to filter by (default: all namespaces). 

237 monthly: Return monthly stats for 3 years if True, daily stats for 3 months if False. 

238 

239 Returns: 

240 NamespaceStats with projects_updated and projects_created histograms. 

241 """ 

242 if monthly: 

243 number_of_month = 12 * 3 

244 else: 

245 number_of_month = 3 

246 today_date = datetime.today().date() + timedelta(days=1) 

247 three_month_ago = today_date - timedelta(days=number_of_month * 30 + 1) 

248 statement_last_update = select(Projects.last_update_date).filter( 

249 Projects.last_update_date.between(three_month_ago, today_date) 

250 ) 

251 statement_create_date = select(Projects.submission_date).filter( 

252 Projects.submission_date.between(three_month_ago, today_date) 

253 ) 

254 if namespace: 

255 statement_last_update = statement_last_update.where( 

256 Projects.namespace == namespace 

257 ) 

258 statement_create_date = statement_create_date.where( 

259 Projects.namespace == namespace 

260 ) 

261 

262 with Session(self._sa_engine) as session: 

263 update_results = session.execute(statement_last_update).all() 

264 create_results = session.execute(statement_create_date).all() 

265 

266 if not update_results: 

267 raise NamespaceNotFoundError( 

268 f"Namespace {namespace} not found in the database" 

269 ) 

270 

271 if monthly: 

272 year_month_str_submission = [ 

273 dt.submission_date.strftime("%Y-%m") for dt in create_results 

274 ] 

275 year_month_str_last_update = [ 

276 dt.last_update_date.strftime("%Y-%m") for dt in update_results 

277 ] 

278 else: 

279 year_month_str_submission = [ 

280 dt.submission_date.strftime("%Y-%m-%d") for dt in create_results 

281 ] 

282 year_month_str_last_update = [ 

283 dt.last_update_date.strftime("%Y-%m-%d") for dt in update_results 

284 ] 

285 

286 counts_submission = dict(Counter(year_month_str_submission)) 

287 counts_last_update = dict(Counter(year_month_str_last_update)) 

288 

289 return NamespaceStats( 

290 namespace=namespace, 

291 projects_updated=counts_last_update, 

292 projects_created=counts_submission, 

293 ) 

294 

295 def upload_tar_info(self, tar_info: TarNamespaceModel) -> None: 

296 """Upload metadata for a namespace tar archive. 

297 

298 Args: 

299 tar_info: Tar archive metadata. 

300 """ 

301 

302 with Session(self._sa_engine) as session: 

303 new_tar = TarNamespace( 

304 file_path=tar_info.file_path, 

305 namespace=tar_info.namespace, 

306 creation_date=tar_info.creation_date, 

307 number_of_projects=tar_info.number_of_projects, 

308 file_size=tar_info.file_size, 

309 ) 

310 session.add(new_tar) 

311 session.commit() 

312 

313 _LOGGER.info("Geo tar info was uploaded successfully!") 

314 

315 def get_tar_info(self, namespace: str) -> TarNamespaceModelReturn: 

316 """Get metadata for namespace tar archives. 

317 

318 Args: 

319 namespace: Namespace of the tar files. 

320 

321 Returns: 

322 TarNamespaceModelReturn with count and list of archive metadata. 

323 """ 

324 

325 with Session(self._sa_engine) as session: 

326 tar_info = session.scalars( 

327 select(TarNamespace) 

328 .where(TarNamespace.namespace == namespace) 

329 .order_by(TarNamespace.creation_date.desc()) 

330 ) 

331 

332 results = [] 

333 for result in tar_info: 

334 results.append( 

335 TarNamespaceModel( 

336 identifier=result.id, 

337 namespace=result.namespace, 

338 file_path=result.file_path, 

339 creation_date=result.creation_date, 

340 number_of_projects=result.number_of_projects, 

341 file_size=result.file_size, 

342 ) 

343 ) 

344 

345 return TarNamespaceModelReturn(count=len(results), results=results) 

346 

347 def delete_tar_info(self, namespace: str = None) -> None: 

348 """Delete tar archive metadata for a namespace. 

349 

350 Args: 

351 namespace: Namespace to delete archives for (default: all namespaces). 

352 """ 

353 

354 with Session(self._sa_engine) as session: 

355 delete_statement = delete(TarNamespace) 

356 if namespace: 

357 delete_statement = delete_statement.where( 

358 TarNamespace.namespace == namespace 

359 ) 

360 session.execute(delete_statement) 

361 session.commit() 

362 _LOGGER.info("Geo tar info was deleted successfully!")