Coverage for middle_layer/review/application_layer/orm_repositories/conflict_declaration.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 import or_ 

19from sqlalchemy.orm import Session 

20 

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

22from propose.domain_layer.entities.proposal import Proposal 

23from review.domain_layer.entities.conflict_declaration import ConflictDeclaration 

24from review.domain_layer.entities.individual_science_review import IndividualScienceReview 

25from review.domain_layer.entities.science_reviewer import ScienceReviewer 

26from review.domain_layer.repositories.conflict_declaration import ConflictDeclarationRepository 

27 

28 

29class ConflictDeclarationORMRepository(ConflictDeclarationRepository): 

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

31 self.session = session 

32 

33 def by_id(self, conflict_declaration_id: int) -> ConflictDeclaration: 

34 return get_object_by_id( 

35 self.session, conflict_declaration_id, ConflictDeclaration, ConflictDeclaration.conflict_declaration_id 

36 ) 

37 

38 def by_isr_id(self, isr_id: int) -> ConflictDeclaration | None: 

39 return get_object_by_id( 

40 self.session, isr_id, ConflictDeclaration, ConflictDeclaration.individual_science_review_id 

41 ) 

42 

43 def by_reviewer_id_and_proposal_id(self, reviewer_id: int, proposal_id: int) -> ConflictDeclaration | None: 

44 return ( 

45 self.session.query(ConflictDeclaration) 

46 .join(IndividualScienceReview) 

47 .join(ScienceReviewer) 

48 .filter(ScienceReviewer.science_reviewer_id == reviewer_id) 

49 .join(Proposal) 

50 .filter(Proposal.proposal_id == proposal_id) 

51 .one_or_none() 

52 ) 

53 

54 def list_all(self) -> list[ConflictDeclaration]: 

55 return list_entities(self.session, ConflictDeclaration, ConflictDeclaration.conflict_declaration_id) 

56 

57 def add(self, conflict_declaration: ConflictDeclaration) -> int: 

58 add_entity(self.session, conflict_declaration) 

59 return conflict_declaration.conflict_declaration_id 

60 

61 def update(self, conflict_declaration: ConflictDeclaration) -> None: 

62 c = self.by_id(conflict_declaration.conflict_declaration_id) 

63 c.individual_science_review_id = conflict_declaration.individual_science_review_id 

64 c.conflict_state = conflict_declaration.conflict_state 

65 c.reason = conflict_declaration.reason 

66 self.session.flush() 

67 

68 def delete(self, conflict_declaration: ConflictDeclaration) -> None: 

69 self.session.delete(conflict_declaration) 

70 self.session.flush() 

71 

72 def list_by_proposal(self, proposal_id: int) -> list[ConflictDeclaration]: 

73 return ( 

74 self.session.query(ConflictDeclaration) 

75 .join(IndividualScienceReview) 

76 .filter(IndividualScienceReview.proposal_id == proposal_id) 

77 .order_by(ConflictDeclaration.conflict_declaration_id) 

78 .all() 

79 ) 

80 

81 def list_by_proposal_with_state_unknown_or_available(self, proposal_id: int) -> list[ConflictDeclaration]: 

82 return ( 

83 self.session.query(ConflictDeclaration) 

84 .join(IndividualScienceReview) 

85 .filter(IndividualScienceReview.proposal_id == proposal_id) 

86 .filter( 

87 or_(ConflictDeclaration.conflict_state == "Unknown", ConflictDeclaration.conflict_state == "Available") 

88 ) 

89 .order_by(ConflictDeclaration.conflict_declaration_id) 

90 .all() 

91 ) 

92 

93 def list_by_science_reviewer(self, science_reviewer_id: int) -> list[ConflictDeclaration]: 

94 return list( 

95 self.session.query(ConflictDeclaration) 

96 .join(IndividualScienceReview) 

97 .filter(IndividualScienceReview.science_reviewer_id == science_reviewer_id) 

98 .order_by(ConflictDeclaration.conflict_declaration_id) 

99 .all() 

100 ) 

101 

102 def list_by_solicitation(self, solicitation_id: int) -> list[ConflictDeclaration]: 

103 return ( 

104 self.session.query(ConflictDeclaration) 

105 .join(IndividualScienceReview) 

106 .join(Proposal) 

107 .filter(Proposal.solicitation_id == solicitation_id) 

108 .order_by(ConflictDeclaration.conflict_declaration_id) 

109 .all() 

110 )