Coverage for src/worblehat/queries/borrowing_queue.py: 24%

1from datetime import datetime 

2 

3from sqlalchemy import exists, select 

4from sqlalchemy.orm import Session 

5 

6from worblehat.models import BookcaseItem, BorrowingLog, QueueEventType, QueueLog, QueuePosition 

7from worblehat.queries.borrowing import has_active_borrowing 

8 

9 

10def list_all_queue_items(sql_session: Session) -> list[QueuePosition]: 

11 return list( 

12 sql_session.scalars( 

13 select(QueuePosition).order_by( 

14 QueuePosition.entered_queue_time, 

15 ), 

16 ).all(), 

17 ) 

18 

19 

20def get_queue_position( 

21 sql_session: Session, 

22 username: str, 

23 item: BookcaseItem, 

24) -> QueuePosition | None: 

25 return sql_session.scalars( 

26 select(QueuePosition).where( 

27 QueuePosition.username == username, 

28 QueuePosition.fk_bookcase_item_uid == item.uid, 

29 ), 

30 ).one_or_none() 

31 

32 

33def is_in_borrowing_queue(sql_session: Session, username: str, item: BookcaseItem) -> bool: 

34 return sql_session.scalar( 

35 select( 

36 exists().where( 

37 QueuePosition.username == username, 

38 QueuePosition.fk_bookcase_item_uid == item.uid, 

39 ), 

40 ), 

41 ) 

42 

43 

44def list_queue_positions_for_item( 

45 sql_session: Session, 

46 item: BookcaseItem, 

47) -> list[QueuePosition]: 

48 return list( 

49 sql_session.scalars( 

50 select(QueuePosition) 

51 .where(QueuePosition.fk_bookcase_item_uid == item.uid) 

52 .order_by(QueuePosition.entered_queue_time), 

53 ).all(), 

54 ) 

55 

56 

57def list_pending_queue_items_for_item( 

58 sql_session: Session, 

59 item: BookcaseItem, 

60) -> list[QueuePosition]: 

61 return list( 

62 sql_session.scalars( 

63 select(QueuePosition) 

64 .where( 

65 QueuePosition.fk_bookcase_item_uid == item.uid, 

66 QueuePosition.notified_available_time.is_(None), 

67 ) 

68 .order_by(QueuePosition.entered_queue_time), 

69 ).all(), 

70 ) 

71 

72 

73def list_queue_log_for_item(sql_session: Session, item: BookcaseItem) -> list[QueueLog]: 

74 return list( 

75 sql_session.scalars( 

76 select(QueueLog) 

77 .where(QueueLog.fk_bookcase_item_uid == item.uid) 

78 .order_by(QueueLog.uid), 

79 ).all(), 

80 ) 

81 

82 

83def join_borrowing_queue( 

84 sql_session: Session, 

85 username: str, 

86 item: BookcaseItem, 

87 _unsafe: bool = False, 

88) -> QueuePosition: 

89 if not _unsafe: 

90 if has_active_borrowing(sql_session, username, item): 

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

92 if is_in_borrowing_queue(sql_session, username, item): 

93 raise ValueError(f"{username} is already queued for this item") 

94 

95 sql_session.add(QueueLog(username, item, QueueEventType.JOINED)) 

96 sql_session.flush() 

97 return sql_session.get_one(QueuePosition, (item.uid, username)) 

98 

99 

100def leave_borrowing_queue( 

101 sql_session: Session, 

102 position: QueuePosition, 

103 _unsafe: bool = False, 

104) -> None: 

105 if not _unsafe and not is_in_borrowing_queue(sql_session, position.username, position.item): 

106 raise ValueError(f"{position.username} is not currently queued for this item") 

107 

108 sql_session.add(QueueLog(position.username, position.item, QueueEventType.LEFT)) 

109 sql_session.flush() 

110 sql_session.expunge(position) 

111 

112 

113def claim_borrowing_queue_position( 

114 sql_session: Session, 

115 position: QueuePosition, 

116 borrowing_log_entry: BorrowingLog, 

117 _unsafe: bool = False, 

118) -> None: 

119 if not _unsafe: 

120 if not is_in_borrowing_queue(sql_session, position.username, position.item): 

121 raise ValueError(f"{position.username} is not currently queued for this item") 

122 if ( 

123 borrowing_log_entry.username != position.username 

124 or borrowing_log_entry.fk_bookcase_item_uid != position.fk_bookcase_item_uid 

125 ): 

126 raise ValueError("borrowing_log_entry does not belong to this queue position") 

127 

128 sql_session.add( 

129 QueueLog( 

130 position.username, 

131 position.item, 

132 QueueEventType.CLAIMED, 

133 borrowing_log_entry=borrowing_log_entry, 

134 ), 

135 ) 

136 sql_session.flush() 

137 sql_session.expunge(position) 

138 

139 

140def expire_borrowing_queue_position( 

141 sql_session: Session, 

142 position: QueuePosition, 

143 _unsafe: bool = False, 

144) -> None: 

145 if not _unsafe and not is_in_borrowing_queue(sql_session, position.username, position.item): 

146 raise ValueError(f"{position.username} is not currently queued for this item") 

147 

148 sql_session.add(QueueLog(position.username, position.item, QueueEventType.EXPIRED)) 

149 sql_session.flush() 

150 sql_session.expunge(position) 

151 

152 

153def notify_borrowing_queue_position( 

154 sql_session: Session, 

155 position: QueuePosition, 

156 notified_at: datetime | None = None, 

157 _unsafe: bool = False, 

158) -> QueuePosition: 

159 if not _unsafe: 

160 if not is_in_borrowing_queue(sql_session, position.username, position.item): 

161 raise ValueError(f"{position.username} is not currently queued for this item") 

162 if position.notified_available_time is not None: 

163 raise ValueError(f"{position.username}'s queue position has already been notified") 

164 

165 sql_session.add( 

166 QueueLog( 

167 position.username, 

168 position.item, 

169 QueueEventType.NOTIFIED, 

170 timestamp=notified_at, 

171 ), 

172 ) 

173 sql_session.flush() 

174 sql_session.refresh(position) 

175 return position