Coverage for middle_layer/review/application_layer/orm_repositories/individual_science_review.py: 100.00%

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# 

18from sqlalchemy.orm import Session 

19 

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

21from propose.domain_layer.entities.proposal import Proposal 

22from review.domain_layer.entities.individual_science_review import IndividualScienceReview 

23from review.domain_layer.entities.science_review_panel import ScienceReviewPanel 

24from review.domain_layer.entities.science_reviewer import ScienceReviewer 

25from review.domain_layer.repositories.individual_science_review import IndividualScienceReviewRepository 

26from solicit.domain_layer.entities.solicitation import Solicitation 

27 

28 

29class IndividualScienceReviewORMRepository(IndividualScienceReviewRepository): 

30 def __init__(self, session: Session): 

31 self.session = session 

32 

33 def by_id(self, isr_id: int) -> IndividualScienceReview: 

34 return get_object_by_id( 

35 self.session, isr_id, IndividualScienceReview, IndividualScienceReview.individual_science_review_id 

36 ) 

37 

38 def list_by_reviewer_id(self, reviewer_id: int) -> list[IndividualScienceReview]: 

39 isrs = list( 

40 self.session.query(IndividualScienceReview) 

41 .join(ScienceReviewer) 

42 .filter(ScienceReviewer.science_reviewer_id == reviewer_id) 

43 .order_by(IndividualScienceReview.individual_science_review_id) 

44 .all() 

45 ) 

46 if len(isrs) == 0: 

47 # Raise a ValueError if no Reviewer exists with the given science_reviewer_id 

48 get_object_by_id(self.session, reviewer_id, ScienceReviewer, ScienceReviewer.science_reviewer_id) 

49 return isrs 

50 

51 def list_by_srp_id(self, srp_id: int) -> list[IndividualScienceReview]: 

52 isrs = ( 

53 self.session.query(IndividualScienceReview) 

54 .join(ScienceReviewPanel.proposals) 

55 .filter(Proposal.proposal_id == IndividualScienceReview.proposal_id) 

56 .filter(ScienceReviewPanel.science_review_panel_id == srp_id) 

57 .order_by(IndividualScienceReview.proposal_id) 

58 .order_by(IndividualScienceReview.science_reviewer_id) 

59 .all() 

60 ) 

61 if not isrs: 

62 # Raise ValueError if no SRP exists with ID srp_id 

63 get_object_by_id(self.session, srp_id, ScienceReviewPanel, ScienceReviewPanel.science_review_panel_id) 

64 return isrs 

65 

66 def list_by_user_id(self, user_id: int) -> list[IndividualScienceReview]: 

67 return ( 

68 self.session.query(IndividualScienceReview) 

69 .join(ScienceReviewer) 

70 .filter(ScienceReviewer.user_id == user_id) 

71 .all() 

72 ) 

73 

74 def by_reviewer_id_and_proposal_code(self, reviewer_id: int, proposal_code: str) -> IndividualScienceReview | None: 

75 return ( 

76 self.session.query(IndividualScienceReview) 

77 .join(ScienceReviewer) 

78 .filter(ScienceReviewer.science_reviewer_id == reviewer_id) 

79 .join(Proposal) 

80 .filter(Proposal.proposal_code == proposal_code) 

81 .one_or_none() 

82 ) 

83 

84 def list_by_solicitation_id(self, solicitation_id: int) -> list[IndividualScienceReview]: 

85 isrs = list( 

86 self.session.query(IndividualScienceReview) 

87 .join(Proposal) 

88 .join(Solicitation) 

89 .filter(Solicitation.solicitation_id == solicitation_id) 

90 .all() 

91 ) 

92 if len(isrs) == 0: 

93 # Raise a ValueError if no Solicitation exists with the given solicitation_id 

94 get_object_by_id(self.session, solicitation_id, Solicitation, Solicitation.solicitation_id) 

95 return isrs 

96 

97 def list_by_proposal_id(self, proposal_id: int) -> list[IndividualScienceReview]: 

98 isrs = ( 

99 self.session.query(IndividualScienceReview).filter(IndividualScienceReview.proposal_id == proposal_id).all() 

100 ) 

101 if not isrs: 

102 # Raise ValueError if no Proposal exists with ID proposal_id 

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

104 return isrs 

105 

106 def list_all(self) -> list[IndividualScienceReview]: 

107 return list_entities( 

108 self.session, IndividualScienceReview, IndividualScienceReview.individual_science_review_id 

109 ) 

110 

111 def add(self, isr: IndividualScienceReview) -> int: 

112 add_entity(self.session, isr) 

113 return isr.individual_science_review_id 

114 

115 def update(self, isr: IndividualScienceReview) -> None: 

116 r = self.by_id(isr.individual_science_review_id) 

117 r.science_reviewer_id = isr.science_reviewer_id 

118 r.proposal_id = isr.proposal_id 

119 r.review_state = isr.review_state 

120 r.review_type = isr.review_type 

121 r.comments_for_the_srp = isr.comments_for_the_srp 

122 r.individual_score = isr.individual_score 

123 r.normalized_score = isr.normalized_score 

124 r.finalized_normalized_score = isr.finalized_normalized_score 

125 r.conflict_declaration = isr.conflict_declaration 

126 self.session.flush() 

127 

128 def delete(self, isr: IndividualScienceReview) -> None: 

129 self.session.delete(isr) 

130 self.session.flush()