Coverage for pepdbagent/modules/view.py: 87%
128 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-01 22:07 +0000
1# View of the PEP. In other words, it is a part of the PEP, or subset of the samples in the PEP.
3import logging
5import peprs
6from sqlalchemy import and_, delete, select
7from sqlalchemy.exc import IntegrityError
8from sqlalchemy.orm import Session
10from pepdbagent.const import DEFAULT_TAG, PKG_NAME
11from pepdbagent.db_utils import (
12 BaseEngine,
13 Projects,
14 Samples,
15 Views,
16 ViewSampleAssociation,
17)
18from pepdbagent.exceptions import (
19 ProjectNotFoundError,
20 SampleAlreadyInView,
21 SampleNotFoundError,
22 SampleNotInViewError,
23 ViewAlreadyExistsError,
24 ViewNotFoundError,
25)
26from pepdbagent.models import CreateViewDictModel, ProjectViews, ViewAnnotation
28_LOGGER = logging.getLogger(PKG_NAME)
31class PEPDatabaseView:
32 """
33 Class that represents Project in Database.
35 While using this class, user can create, retrieve, delete, and update projects from database
36 """
38 def __init__(self, pep_db_engine: BaseEngine):
39 """
40 Args:
41 pep_db_engine: PEPDatabaseAgent engine object.
42 """
43 self._sa_engine = pep_db_engine.engine
44 self._pep_db_engine = pep_db_engine
46 def get(
47 self,
48 namespace: str,
49 name: str,
50 tag: str = DEFAULT_TAG,
51 view_name: str = None,
52 raw: bool = True,
53 ) -> peprs.Project | dict | None:
54 """Retrieve a view of the project from the database.
56 A view is a subset of samples in the project.
58 Args:
59 namespace: Namespace of the project.
60 name: Name of the project.
61 tag: Tag of the project.
62 view_name: Name of the view.
63 raw: Return raw dict if True, peprs.Project object if False.
65 Returns:
66 Raw dict or peprs.Project with only the view's samples.
68 Raises:
69 ViewNotFoundError: If the view does not exist.
70 """
71 _LOGGER.debug(f"Get view {view_name} from {namespace}/{name}:{tag}")
72 view_statement = select(Views).where(
73 and_(
74 Views.project_mapping.has(namespace=namespace, name=name, tag=tag),
75 Views.name == view_name,
76 )
77 )
79 with Session(self._sa_engine) as sa_session:
80 view = sa_session.scalar(view_statement)
81 if not view:
82 raise ViewNotFoundError(
83 f"View {view_name} of the project {namespace}/{name}:{tag} does not exist"
84 )
85 samples = [sample.sample.sample for sample in view.samples]
86 config = view.project_mapping.config
87 sub_project_dict = {"config": config, "samples": samples}
88 if raw:
89 return sub_project_dict
90 else:
91 return peprs.Project.from_dict(sub_project_dict)
93 def get_annotation(
94 self, namespace: str, name: str, tag: str = DEFAULT_TAG, view_name: str = None
95 ) -> ViewAnnotation:
96 """Get annotation for a view.
98 Args:
99 namespace: Namespace of the project.
100 name: Name of the project.
101 tag: Tag of the project.
102 view_name: Name of the view.
104 Returns:
105 ViewAnnotation with project coordinates, name, description, and sample count.
107 Raises:
108 ViewNotFoundError: If the view does not exist.
109 """
110 _LOGGER.debug(
111 f"Get annotation for view {view_name} in {namespace}/{name}:{tag}"
112 )
113 view_statement = select(Views).where(
114 and_(
115 Views.project_mapping.has(namespace=namespace, name=name, tag=tag),
116 Views.name == view_name,
117 )
118 )
120 with Session(self._sa_engine) as sa_session:
121 view = sa_session.scalar(view_statement)
122 if not view:
123 raise ViewNotFoundError(
124 f"View {name} of the project {namespace}/{name}:{tag} does not exist"
125 )
126 return ViewAnnotation(
127 project_namespace=namespace,
128 project_name=name,
129 project_tag=tag,
130 name=view.name,
131 description=view.description,
132 number_of_samples=len(view.samples),
133 )
135 def create(
136 self,
137 view_name: str,
138 view_dict: dict | CreateViewDictModel,
139 description: str = None,
140 no_fail: bool = False,
141 ) -> None:
142 """Create a view of a project in the database.
144 Args:
145 view_name: Name for the new view.
146 view_dict: View definition with project_namespace, project_name, project_tag,
147 and sample_list keys (or a CreateViewDictModel).
148 description: Optional description of the view.
149 no_fail: Skip samples that do not exist instead of raising an error.
151 Raises:
152 ProjectNotFoundError: If the project does not exist.
153 SampleNotFoundError: If a sample does not exist and no_fail is False.
154 ViewAlreadyExistsError: If a view with the same name already exists.
155 """
156 _LOGGER.debug(
157 f"Creating view {view_name} with provided info: (view_dict: {view_dict})"
158 )
159 if isinstance(view_dict, dict):
160 view_dict = CreateViewDictModel(**view_dict)
162 project_statement = select(Projects).where(
163 and_(
164 Projects.namespace == view_dict.project_namespace,
165 Projects.name == view_dict.project_name,
166 Projects.tag == view_dict.project_tag,
167 )
168 )
169 try:
170 with Session(self._sa_engine) as sa_session:
171 project = sa_session.scalar(project_statement)
172 if not project:
173 raise ProjectNotFoundError(
174 f"Project {view_dict.project_namespace}/{view_dict.project_name}:{view_dict.project_tag} does not exist"
175 )
176 view = Views(
177 name=view_name,
178 description=description,
179 project_mapping=project,
180 )
181 sa_session.add(view)
183 for sample_name in view_dict.sample_list:
184 sample_statement = select(Samples.id).where(
185 and_(
186 Samples.project_id == project.id,
187 Samples.sample_name == sample_name,
188 )
189 )
190 sample_id_tuple = sa_session.execute(sample_statement).one_or_none()
191 if sample_id_tuple:
192 sample_id = sample_id_tuple[0]
193 elif not sample_id_tuple and not no_fail:
194 raise SampleNotFoundError(
195 f"Sample {view_dict.project_namespace}/{view_dict.project_name}:{view_dict.project_tag}:{sample_name} does not exist"
196 )
197 else:
198 continue
200 sa_session.add(
201 ViewSampleAssociation(sample_id=sample_id, view=view)
202 )
204 sa_session.commit()
205 except IntegrityError:
206 raise ViewAlreadyExistsError(
207 f"View {view_name} of the project {view_dict.project_namespace}/{view_dict.project_name}:{view_dict.project_tag} already exists"
208 )
210 def delete(
211 self,
212 project_namespace: str,
213 project_name: str,
214 project_tag: str = DEFAULT_TAG,
215 view_name: str = None,
216 ) -> None:
217 """Delete a view from the database.
219 Args:
220 project_namespace: Namespace of the project.
221 project_name: Name of the project.
222 project_tag: Tag of the project.
223 view_name: Name of the view to delete.
225 Raises:
226 ViewNotFoundError: If the view does not exist.
227 """
228 _LOGGER.debug(
229 f"Deleting view {view_name} from {project_namespace}/{project_name}:{project_tag}"
230 )
231 view_statement = select(Views).where(
232 and_(
233 Views.project_mapping.has(
234 namespace=project_namespace, name=project_name, tag=project_tag
235 ),
236 Views.name == view_name,
237 )
238 )
240 with Session(self._sa_engine) as sa_session:
241 view = sa_session.scalar(view_statement)
242 if not view:
243 raise ViewNotFoundError(
244 f"View {view_name} of the project {project_namespace}/{project_name}:{project_tag} does not exist"
245 )
246 sa_session.delete(view)
247 sa_session.commit()
249 def add_sample(
250 self,
251 namespace: str,
252 name: str,
253 tag: str,
254 view_name: str,
255 sample_name: str | list[str],
256 ) -> None:
257 """Add one or more samples to a view.
259 Args:
260 namespace: Namespace of the project.
261 name: Name of the project.
262 tag: Tag of the project.
263 view_name: Name of the view.
264 sample_name: Sample name or list of sample names.
266 Raises:
267 ViewNotFoundError: If the view does not exist.
268 SampleNotFoundError: If a sample does not exist.
269 SampleAlreadyInView: If a sample is already in the view.
270 """
271 _LOGGER.debug(
272 f"Adding sample {sample_name} to view {view_name} in {namespace}/{name}:{tag}"
273 )
274 if isinstance(sample_name, str):
275 sample_name = [sample_name]
276 view_statement = select(Views).where(
277 and_(
278 Views.project_mapping.has(namespace=namespace, name=name, tag=tag),
279 Views.name == view_name,
280 )
281 )
282 try:
283 with Session(self._sa_engine) as sa_session:
284 view = sa_session.scalar(view_statement)
285 if not view:
286 raise ViewNotFoundError(
287 f"View {view_name} of the project {namespace}/{name}:{tag} does not exist"
288 )
289 for sample_name_one in sample_name:
290 sample_statement = select(Samples).where(
291 and_(
292 Samples.project_id == view.project_mapping.id,
293 Samples.sample_name == sample_name_one,
294 )
295 )
296 sample = sa_session.scalar(sample_statement)
297 if not sample:
298 raise SampleNotFoundError(
299 f"Sample {namespace}/{name}:{tag}:{sample_name} does not exist"
300 )
302 sa_session.add(ViewSampleAssociation(sample=sample, view=view))
303 sa_session.commit()
304 except IntegrityError:
305 raise SampleAlreadyInView(
306 f"Sample {namespace}/{name}:{tag}:{sample_name} already in view {view_name}"
307 )
309 def remove_sample(
310 self,
311 namespace: str,
312 name: str,
313 tag: str,
314 view_name: str,
315 sample_name: str,
316 ) -> None:
317 """Remove a sample from a view.
319 Args:
320 namespace: Namespace of the project.
321 name: Name of the project.
322 tag: Tag of the project.
323 view_name: Name of the view.
324 sample_name: Name of the sample to remove.
326 Raises:
327 ViewNotFoundError: If the view does not exist.
328 SampleNotInViewError: If the sample is not in the view.
329 """
330 _LOGGER.debug(
331 f"Removing sample {sample_name} from view {view_name} in {namespace}/{name}:{tag}"
332 )
333 view_statement = select(Views).where(
334 and_(
335 Views.project_mapping.has(namespace=namespace, name=name, tag=tag),
336 Views.name == view_name,
337 )
338 )
340 with Session(self._sa_engine) as sa_session:
341 view = sa_session.scalar(view_statement)
342 if not view:
343 raise ViewNotFoundError(
344 f"View {view_name} of the project {namespace}/{name}:{tag} does not exist"
345 )
346 sample_statement = select(Samples).where(
347 and_(
348 Samples.project_id == view.project_mapping.id,
349 Samples.sample_name == sample_name,
350 )
351 )
352 sample = sa_session.scalar(sample_statement)
353 if sample.id not in [view_sample.sample_id for view_sample in view.samples]:
354 raise SampleNotInViewError(
355 f"Sample {namespace}/{name}:{tag}:{sample_name} does not exist in view {view_name}"
356 )
357 delete_statement = delete(ViewSampleAssociation).where(
358 and_(
359 ViewSampleAssociation.sample_id == sample.id,
360 ViewSampleAssociation.view_id == view.id,
361 )
362 )
363 sa_session.execute(delete_statement)
364 sa_session.commit()
366 def get_snap_view(
367 self,
368 namespace: str,
369 name: str,
370 tag: str,
371 sample_name_list: list[str],
372 raw: bool = False,
373 ) -> peprs.Project | dict:
374 """Get an ephemeral view of a project limited to a set of samples.
376 The snap view is not persisted in the database.
378 Args:
379 namespace: Namespace of the project.
380 name: Name of the project.
381 tag: Tag of the project.
382 sample_name_list: Sample names to include, e.g., ["sample1", "sample2"].
383 raw: Return raw dict if True, peprs.Project object if False.
385 Returns:
386 Raw dict or peprs.Project containing only the requested samples.
388 Raises:
389 ProjectNotFoundError: If the project does not exist.
390 SampleNotFoundError: If any sample does not exist.
391 """
392 _LOGGER.debug(f"Creating snap view for {namespace}/{name}:{tag}")
393 project_statement = select(Projects).where(
394 and_(
395 Projects.namespace == namespace,
396 Projects.name == name,
397 Projects.tag == tag,
398 )
399 )
400 with Session(self._sa_engine) as sa_session:
401 project = sa_session.scalar(project_statement)
402 if not project:
403 raise ProjectNotFoundError(
404 f"Project {namespace}/{name}:{tag} does not exist"
405 )
406 samples = []
407 for sample_name in sample_name_list:
408 sample_statement = select(Samples).where(
409 and_(
410 Samples.project_id == project.id,
411 Samples.sample_name == sample_name,
412 )
413 )
414 sample = sa_session.scalar(sample_statement)
415 if not sample:
416 raise SampleNotFoundError(
417 f"Sample {namespace}/{name}:{tag}:{sample_name} does not exist"
418 )
419 samples.append(sample.sample)
420 config = project.config
422 if raw:
423 return {"config": config, "samples": samples}
424 else:
425 return peprs.Project.from_dict({"config": config, "samples": samples})
427 def get_views_annotation(
428 self, namespace: str, name: str, tag: str = DEFAULT_TAG
429 ) -> ProjectViews | None:
430 """Get annotation for all views of a project.
432 Args:
433 namespace: Namespace of the project.
434 name: Name of the project.
435 tag: Tag of the project.
437 Returns:
438 ProjectViews with a list of ViewAnnotation objects.
439 """
440 _LOGGER.debug(f"Get views annotation for {namespace}/{name}:{tag}")
441 statement = select(Views).where(
442 Views.project_mapping.has(namespace=namespace, name=name, tag=tag),
443 )
444 views_list = []
446 with Session(self._sa_engine) as session:
447 views = session.scalars(statement)
448 for view in views:
449 views_list.append(
450 ViewAnnotation(
451 name=view.name,
452 description=view.description,
453 number_of_samples=len(view.samples),
454 )
455 )
457 return ProjectViews(namespace=namespace, name=name, tag=tag, views=views_list)