Coverage for middle_layer/review/application_layer/orm_repositories/science_review_panel.py: 97.62%

1# Copyright 2024 Associated Universities, Inc. 

2# 

3# This file is part of Telescope Time Allocation Tools (TTAT). 

4# 

5# TTAT is free software: you can redistribute it and/or modify 

6# it under the terms of the GNU General Public License as published by 

7# the Free Software Foundation, either version 3 of the License, or 

8# any later version. 

9# 

10# TTAT is distributed in the hope that it will be useful, 

11# but WITHOUT ANY WARRANTY; without even the implied warranty of 

12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

13# GNU General Public License for more details. 

14# 

15# You should have received a copy of the GNU General Public License 

16# along with TTAT. If not, see <https://www.gnu.org/licenses/>. 

17 

18"""ORMRepository for ScienceReviewPanel""" 

19 

20 

21from sqlalchemy.orm.session import Session 

22 

23from common.application_layer.orm_repositories import add_entity, get_object_by_id, list_entities 

24from propose.domain_layer.entities.proposal import Proposal 

25from review.domain_layer.entities.individual_science_review import IndividualScienceReview 

26from review.domain_layer.entities.science_review_panel import ScienceReviewPanel 

27from review.domain_layer.entities.science_reviewer import ScienceReviewer 

28from review.domain_layer.repositories.science_review_panel import ScienceReviewPanelRepository 

29 

30 

31class ScienceReviewPanelORMRepository(ScienceReviewPanelRepository): 

32 def __init__(self, session: Session) -> None: 

33 self.session = session 

34 

35 def add(self, science_review_panel: ScienceReviewPanel) -> int: 

36 add_entity(self.session, science_review_panel) 

37 return science_review_panel.science_review_panel_id 

38 

39 def by_id(self, science_review_panel_id: int) -> ScienceReviewPanel: 

40 return get_object_by_id( 

41 self.session, science_review_panel_id, ScienceReviewPanel, ScienceReviewPanel.science_review_panel_id 

42 ) 

43 

44 def list_all(self) -> list[ScienceReviewPanel]: 

45 return list_entities(self.session, ScienceReviewPanel, ScienceReviewPanel.science_review_panel_id) 

46 

47 def list_by_solicitation_id(self, solicitation_id: int) -> list[ScienceReviewPanel]: 

48 return ( 

49 self.session.query(ScienceReviewPanel) 

50 .filter(ScienceReviewPanel.solicitation_id == solicitation_id) 

51 .order_by(ScienceReviewPanel.science_review_panel_id) 

52 .all() 

53 ) 

54 

55 def by_proposal_id(self, proposal_id: int) -> ScienceReviewPanel | None: 

56 srp = ( 

57 self.session.query(ScienceReviewPanel) 

58 .filter(ScienceReviewPanel.proposals.any(Proposal.proposal_id == proposal_id)) 

59 .order_by(ScienceReviewPanel.science_review_panel_id) 

60 .one_or_none() 

61 ) 

62 if srp is None: 

63 # Raise ValueError if Proposal doesn't exist 

64 get_object_by_id(self.session, proposal_id, Proposal, Proposal.proposal_id) 

65 return srp 

66 

67 def list_by_science_reviewer_id(self, science_reviewer_id: int) -> list[ScienceReviewPanel]: 

68 srs_srp = ( 

69 self.session.query(ScienceReviewPanel) 

70 .join(ScienceReviewer) 

71 .filter(ScienceReviewer.science_reviewer_id == science_reviewer_id) 

72 .one() 

73 ) 

74 isr_srps = ( 

75 self.session.query(ScienceReviewPanel) 

76 .join(ScienceReviewPanel.proposals) 

77 .join(IndividualScienceReview) 

78 .filter(IndividualScienceReview.science_reviewer_id == science_reviewer_id) 

79 .order_by(ScienceReviewPanel.science_review_panel_id) 

80 .all() 

81 ) 

82 if srs_srp not in isr_srps: 

83 isr_srps.append(srs_srp) 

84 return isr_srps 

85 

86 def update(self, science_review_panel: ScienceReviewPanel) -> None: 

87 srp = self.by_id(science_review_panel.science_review_panel_id) 

88 srp.science_review_panel_name = science_review_panel.science_review_panel_name 

89 srp.solicitation = science_review_panel.solicitation 

90 srp.science_reviewers = science_review_panel.science_reviewers 

91 srp.proposals = science_review_panel.proposals 

92 srp.science_categories = science_review_panel.science_categories 

93 srp.minimum_science_reviewers_per_proposal = science_review_panel.minimum_science_reviewers_per_proposal 

94 self.session.flush() 

95 

96 def delete(self, science_review_panel: ScienceReviewPanel) -> None: 

97 self.session.delete(science_review_panel) 

98 self.session.flush()