Coverage for src/worblehat/queries/borrowing.py: 36%

1from datetime import datetime, timedelta 

2 

3from sqlalchemy import exists, select 

4from sqlalchemy.orm import Session 

5 

6from worblehat.models import BookcaseItem, Borrowing, BorrowingEventType, BorrowingLog 

7 

8DEFAULT_LOAN_DAYS = 30 

9 

10 

11def list_active_borrowings(sql_session: Session) -> list[Borrowing]: 

12 return list( 

13 sql_session.scalars( 

14 select(Borrowing).order_by(Borrowing.due_time), 

15 ).all(), 

16 ) 

17 

18 

19def list_active_borrowings_for_item( 

20 sql_session: Session, 

21 item: BookcaseItem, 

22) -> list[Borrowing]: 

23 return list( 

24 sql_session.scalars( 

25 select(Borrowing) 

26 .where(Borrowing.fk_bookcase_item_uid == item.uid) 

27 .order_by(Borrowing.due_time), 

28 ).all(), 

29 ) 

30 

31 

32def get_active_borrowing( 

33 sql_session: Session, 

34 username: str, 

35 item: BookcaseItem, 

36) -> Borrowing | None: 

37 return sql_session.scalars( 

38 select(Borrowing).where( 

39 Borrowing.username == username, 

40 Borrowing.fk_bookcase_item_uid == item.uid, 

41 ), 

42 ).one_or_none() 

43 

44 

45def has_active_borrowing(sql_session: Session, username: str, item: BookcaseItem) -> bool: 

46 return sql_session.scalar( 

47 select( 

48 exists().where( 

49 Borrowing.username == username, 

50 Borrowing.fk_bookcase_item_uid == item.uid, 

51 ), 

52 ), 

53 ) 

54 

55 

56def list_borrowings_for_isbn(sql_session: Session, isbn: str) -> list[Borrowing]: 

57 return list( 

58 sql_session.scalars( 

59 select(Borrowing) 

60 .join( 

61 BookcaseItem, 

62 BookcaseItem.uid == Borrowing.fk_bookcase_item_uid, 

63 ) 

64 .where(BookcaseItem.isbn == isbn) 

65 .order_by(Borrowing.username), 

66 ).all(), 

67 ) 

68 

69 

70def list_overdue_borrowings(sql_session: Session) -> list[Borrowing]: 

71 return list( 

72 sql_session.scalars( 

73 select(Borrowing) 

74 .where(Borrowing.due_time < datetime.now()) 

75 .order_by(Borrowing.due_time), 

76 ).all(), 

77 ) 

78 

79 

80def list_borrowing_log_for_item( 

81 sql_session: Session, 

82 item: BookcaseItem, 

83) -> list[BorrowingLog]: 

84 return list( 

85 sql_session.scalars( 

86 select(BorrowingLog) 

87 .where(BorrowingLog.fk_bookcase_item_uid == item.uid) 

88 .order_by(BorrowingLog.due_time), 

89 ).all(), 

90 ) 

91 

92 

93def borrow_item( 

94 sql_session: Session, 

95 username: str, 

96 item: BookcaseItem, 

97 loan_days: int = DEFAULT_LOAN_DAYS, 

98 _unsafe: bool = False, 

99) -> tuple[Borrowing, BorrowingLog]: 

100 if not _unsafe: 

101 if has_active_borrowing(sql_session, username, item): 

102 raise ValueError(f"{username} already has an active borrowing of this item") 

103 if len(list_active_borrowings_for_item(sql_session, item)) >= item.amount: 

104 raise ValueError("No available copies of this item to borrow") 

105 

106 due_time = datetime.now() + timedelta(days=loan_days) 

107 log_entry = BorrowingLog(username, item, BorrowingEventType.BORROWED, due_time=due_time) 

108 sql_session.add(log_entry) 

109 sql_session.flush() 

110 current = sql_session.get_one(Borrowing, (item.uid, username)) 

111 return current, log_entry 

112 

113 

114def renew_borrowing( 

115 sql_session: Session, 

116 borrowing: Borrowing, 

117 loan_days: int = DEFAULT_LOAN_DAYS, 

118 _unsafe: bool = False, 

119) -> Borrowing: 

120 if ( 

121 not _unsafe 

122 and get_active_borrowing(sql_session, borrowing.username, borrowing.item) is None 

123 ): 

124 raise ValueError(f"{borrowing.username} does not currently have this item borrowed") 

125 

126 due_time = datetime.now() + timedelta(days=loan_days) 

127 sql_session.add( 

128 BorrowingLog( 

129 borrowing.username, 

130 borrowing.item, 

131 BorrowingEventType.RENEWED, 

132 due_time=due_time, 

133 ), 

134 ) 

135 sql_session.flush() 

136 sql_session.refresh(borrowing) 

137 return borrowing 

138 

139 

140def return_item(sql_session: Session, borrowing: Borrowing, _unsafe: bool = False) -> None: 

141 if ( 

142 not _unsafe 

143 and get_active_borrowing(sql_session, borrowing.username, borrowing.item) is None 

144 ): 

145 raise ValueError(f"{borrowing.username} does not currently have this item borrowed") 

146 

147 sql_session.add( 

148 BorrowingLog(borrowing.username, borrowing.item, BorrowingEventType.RETURNED), 

149 ) 

150 sql_session.flush() 

151 sql_session.expunge(borrowing)