Coverage for middle_layer/propose/application_layer/orm_repositories/capability_request.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 datetime import datetime, timezone 

19 

20from sqlalchemy import update 

21from sqlalchemy.orm 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 AllocationRequest, CapabilityRequest, ProposalCopy 

25from propose.domain_layer.repositories.proposal import CapabilityRequestRepository 

26 

27 

28class CapabilityRequestORMRepository(CapabilityRequestRepository): 

29 def __init__(self, session: Session): 

30 self.session = session 

31 

32 def by_id(self, capability_request_id: int) -> CapabilityRequest: 

33 return get_object_by_id( 

34 self.session, capability_request_id, CapabilityRequest, CapabilityRequest.capability_request_id 

35 ) 

36 

37 def list_all(self) -> list[CapabilityRequest]: 

38 return list_entities(self.session, CapabilityRequest, CapabilityRequest.capability_request_id) 

39 

40 def list_by_allocation_request_id(self, ar_id: int) -> list[CapabilityRequest]: 

41 return list( 

42 self.session.query(CapabilityRequest) 

43 .filter(CapabilityRequest.allocation_request_id == ar_id) 

44 .order_by(CapabilityRequest.capability_request_id) 

45 .all() 

46 ) 

47 

48 def add(self, capability_request: CapabilityRequest) -> int: 

49 add_entity(self.session, capability_request) 

50 ar = get_object_by_id( 

51 self.session, 

52 capability_request.allocation_request_id, 

53 AllocationRequest, 

54 AllocationRequest.allocation_request_id, 

55 ) 

56 proposal_copy = get_object_by_id(self.session, ar.proposal_copy_id, ProposalCopy, ProposalCopy.proposal_copy_id) 

57 proposal_copy.modified_timestamp = datetime.now(timezone.utc) 

58 self.session.flush() 

59 return capability_request.capability_request_id 

60 

61 def update(self, capability_request: CapabilityRequest) -> None: 

62 self.session.execute( 

63 update(CapabilityRequest) 

64 .where(CapabilityRequest.capability_request_id == capability_request.capability_request_id) 

65 .values( 

66 solicitation_facility_capability_id=capability_request.solicitation_facility_capability.solicitation_facility_capability_id 

67 ) 

68 ) 

69 ar = get_object_by_id( 

70 self.session, 

71 capability_request.allocation_request_id, 

72 AllocationRequest, 

73 AllocationRequest.allocation_request_id, 

74 ) 

75 proposal_copy = get_object_by_id(self.session, ar.proposal_copy_id, ProposalCopy, ProposalCopy.proposal_copy_id) 

76 proposal_copy.modified_timestamp = datetime.now(timezone.utc) 

77 self.session.flush() 

78 

79 def delete(self, capability_request: CapabilityRequest) -> None: 

80 self.session.delete(capability_request) 

81 self.session.flush()