Coverage for app / services / search_service.py: 69%

1""" 

2This module provides search functionality for books in the database. 

3 

4It includes methods to search books by categories, authors, or titles, with optional 

5filters for user-specific status and feedback. The queries are tailored to return 

6sorted results, refined to meet various search criteria. If a user is authenticated, 

7search results will include personalized status and feedback data. 

8""" 

9from flask_security import current_user 

10from sqlalchemy import asc, and_, exists, select, literal 

11from sqlalchemy.orm import contains_eager, noload, make_transient 

12 

13from app import db 

14from app.models import Book, ReadingStatus, Feedback, TagBook, Tag 

15 

16 

17def search_by_categories(categories, status_filter: str = None, 

18 feedback_filter: str = None, tag_filter: list[str] = None) -> list[Book]: 

19 """ 

20 Searches for books based on categories and optional filters for status and feedback. 

21  

22 This retrieves a list of books matching one or more categories. Results are sorted 

23 alphabetically by title. Optional filters can refine search results by user status  

24 or feedback. 

25 """ 

26 if not categories: 26 ↛ 27line 26 didn't jump to line 27 because the condition on line 26 was never true

27 return [] # Return an empty list if no categories are provided 

28 

29 # Query to search and sort books based on the provided requirements 

30 query = ((db.session.query(Book) 

31 .filter(Book.categories_flat.in_(categories))) # match in one of the categories 

32 .order_by(asc(Book.title))) # sort by title 

33 

34 query = _add_user_status_and_feedback_joins(query) 

35 

36 return _finish_building_query_and_execute(feedback_filter, query, status_filter, tag_filter) 

37 

38 

39def search_by_author(author: str, status_filter: str, 

40 feedback_filter: str, tag_filter: list[str]) -> list[Book]: 

41 """Search for books by author's name.""" 

42 return _search_by_attribute("author", author, status_filter, feedback_filter, tag_filter) 

43 

44 

45def search_by_title(title, status_filter: str, feedback_filter: str, 

46 tag_filter: list[str]) -> list[Book]: 

47 """Search for books by title.""" 

48 return _search_by_attribute("title", title, status_filter, feedback_filter, tag_filter) 

49 

50 

51_VALID_SEARCH_BY_ATTRIBUTES = {"author", "title"} 

52 

53 

54def _search_by_attribute(attribute: str, value: str, status_filter: str = None, 

55 feedback_filter: str = None, tag_filter: list[str] = None) -> list[Book]: 

56 """ 

57 Searches for books in the database by a specified attribute and value. Filters  

58 for status and feedback can also be applied. 

59  

60 This function queries the `Book` model, using a case-insensitive partial  

61 match for the attribute's value. Filters can refine results by book status  

62 or user feedback. When the value is "*", all books are returned sorted  

63 by the attribute. 

64 

65 :return: A list of `Book` instances that meet the filters. An empty list is  

66 returned if no matches are found. 

67 :rtype: list[Book] 

68 :raises ValueError: If the attribute provided is not valid. 

69 """ 

70 if attribute not in _VALID_SEARCH_BY_ATTRIBUTES: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 raise ValueError( 

72 f"Invalid attribute '{attribute}'. Must be one of {_VALID_SEARCH_BY_ATTRIBUTES}." 

73 ) 

74 if not value: 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true

75 return [] 

76 

77 query = _add_user_status_and_feedback_joins(db.session.query(Book)) 

78 

79 # Order by the selected attribute 

80 query = query.order_by(asc(getattr(Book, attribute))) 

81 

82 # Handle the special case for "*" to return all books sorted by the attribute 

83 if value != "*": 

84 # Perform a case-insensitive partial match (using ilike) 

85 query = query.filter(getattr(Book, attribute).ilike(f"%{value}%")) 

86 

87 return _finish_building_query_and_execute(feedback_filter, query, status_filter, tag_filter) 

88 

89 

90def _finish_building_query_and_execute(feedback_filter, query, status_filter, tag_filter): 

91 query = _add_status_and_feedback_filters(query, status_filter, feedback_filter) 

92 # Add the tag filter if provided 

93 query = query.filter(Book.tags.any(Tag.name.in_(tag_filter))) if tag_filter else query 

94 # execute the query 

95 books = query.all() 

96 for book in books: 

97 make_transient(book) 

98 db.session.expire_all() # expire all books to prevent stale data from being returned 

99 return books 

100 

101 

102def _add_status_and_feedback_filters(query, status_filter, feedback_filter): 

103 if status_filter: 103 ↛ 104line 103 didn't jump to line 104 because the condition on line 103 was never true

104 if status_filter != "none": 

105 # handles read and up_next 

106 query = query.filter(ReadingStatus.status == status_filter) 

107 else: 

108 # finds only books without a status set 

109 query = query.filter(ReadingStatus.status.is_(None)) 

110 if feedback_filter: 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true

111 if feedback_filter != "none": 

112 # handles like and dislike 

113 query = query.filter(Feedback.feedback == feedback_filter) 

114 else: 

115 # finds only books without a feedback set 

116 query = query.filter(Feedback.feedback.is_(None)) 

117 return query 

118 

119 

120def _add_user_status_and_feedback_joins(query): 

121 user_id = current_user.id if current_user.is_authenticated else None 

122 if user_id: 122 ↛ 126line 122 didn't jump to line 126 because the condition on line 122 was never true

123 # Define a constant for EXISTS check 

124 # pylint: disable=invalid-name 

125 # noinspection PyPep8Naming 

126 EXISTS_CHECK = literal(1).label('exists_check') 

127 

128 # Define the subquery to check for user's tags 

129 user_tag_exists = ( 

130 select(EXISTS_CHECK) 

131 .select_from(Tag) 

132 .where( 

133 and_( 

134 Tag.id == TagBook.tag_id, 

135 Tag.owner_id == user_id 

136 ) 

137 ) 

138 .correlate(TagBook) 

139 .alias('user_tag_check') 

140 ) 

141 

142 query = ( 

143 query 

144 # Join with ReadingStatus 

145 .outerjoin( 

146 ReadingStatus, 

147 and_( 

148 ReadingStatus.book_id == Book.id, 

149 ReadingStatus.user_id == user_id 

150 ) 

151 ) 

152 # Join with Feedback 

153 .outerjoin( 

154 Feedback, 

155 and_( 

156 Feedback.book_id == Book.id, 

157 Feedback.user_id == user_id 

158 ) 

159 ) 

160 # Join with TagBook using the exists subquery 

161 .outerjoin( 

162 TagBook, 

163 and_( 

164 TagBook.book_id == Book.id, 

165 exists(user_tag_exists) 

166 ) 

167 ) 

168 # Join with Tag 

169 .outerjoin( 

170 Tag, 

171 and_( 

172 Tag.id == TagBook.tag_id, 

173 Tag.owner_id == user_id 

174 ) 

175 ) 

176 .options( 

177 contains_eager(Book.reading_statuses).load_only( 

178 ReadingStatus.id, 

179 ReadingStatus.status 

180 ).noload(ReadingStatus.user), 

181 contains_eager(Book.feedbacks).load_only( 

182 Feedback.id, 

183 Feedback.feedback 

184 ).noload(Feedback.user), 

185 contains_eager(Book.tags).contains_eager(TagBook.tag), 

186 contains_eager(Book.tags).load_only( 

187 TagBook.id, 

188 TagBook.tag_id 

189 ), 

190 contains_eager(Book.tags).contains_eager(TagBook.tag).load_only( 

191 Tag.id, 

192 Tag.name, 

193 Tag.color 

194 ), 

195 contains_eager(Book.tags).noload(TagBook.book), 

196 contains_eager(Book.tags).contains_eager(TagBook.tag).noload(Tag.owner) 

197 ) 

198 ) 

199 else: 

200 # If no user logged in, load no relationships 

201 query = query.options( 

202 noload(Book.reading_statuses), 

203 noload(Book.feedbacks), 

204 noload(Book.tags) 

205 ) 

206 

207 return query