Coverage for app / services / book_service.py: 70%
114 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-06 04:49 +0000
1"""
2Services for working with books
3"""
4# pylint: disable=raise-missing-from
5from sqlalchemy import update, delete
6from sqlalchemy.exc import IntegrityError, InvalidRequestError
7from sqlalchemy.orm import noload, joinedload
9from app import db
10from app.forms import BookForm
11from app.helpers.utilities import sanitize, sanitize_categories_flat
12from app.models import Book, Feedback, ReadingStatus, FeedbackEnum, ReadingStatusEnum
15def get_book_by_id(book_id, user_id=None, load_status=False, load_feedback=False):
16 """
17 Fetches a book entity by its unique identifier and optionally joins related
18 information such as reading statuses and feedback, depending on the provided
19 parameters. This method ensures that the query will only include the relevant
20 records for the given user if `user_id` is supplied. If `load_status` or
21 `load_feedback` is True, it will perform an OUTER JOIN to fetch related data.
22 Otherwise, the relationships will not be loaded.
24 This function is designed to optimize database queries by controlling the
25 loading of related entities, reducing unnecessary data retrieval.
27 :param book_id: Unique identifier of the book to be fetched.
28 :type book_id: int
29 :param user_id: Unique identifier of the user for filtering statuses or feedback
30 (optional, defaults to None).
31 :type user_id: int, optional
32 :param load_status: If True, includes the relevant reading statuses for the
33 specified user. If False, skips the loading of reading statuses. Defaults
34 to False.
35 :type load_status: bool
36 :param load_feedback: If True, includes the relevant feedback for the specified
37 user. If False, skips the loading of feedback. Defaults to False.
38 :type load_feedback: bool
39 :return: A Book instance matching the provided `book_id`, with related reading
40 statuses and feedback if requested, or None if no match is found.
41 :rtype: Book or None
42 """
43 query = db.session.query(Book)
45 if load_status and user_id is not None:
46 # An OUTER JOIN will occur here, loading only related ReadingStatus entries
47 # that match BOTH user_id and book_id. If no matches are found, the relationship
48 # will simply be an empty list.
49 query = query.options(
50 joinedload(Book.reading_statuses.and_(
51 (ReadingStatus.user_id == user_id) & (ReadingStatus.book_id == book_id)
52 ))
53 )
54 else:
55 query = query.options(noload(Book.reading_statuses))
57 if load_feedback and user_id is not None:
58 # Similarly, an OUTER JOIN will occur here for Feedback, following the same logic as above.
59 query = query.options(
60 joinedload(Book.feedbacks.and_(
61 (Feedback.user_id == user_id) & (Feedback.book_id == book_id)
62 ))
63 )
64 else:
65 query = query.options(noload(Book.feedbacks))
67 # If no ReadingStatus or Feedback records exist, the book will still be returned,
68 # but relationships will be empty lists.
69 return query.filter_by(id=book_id).first()
72def add_new_book(book_form: BookForm) -> Book:
73 """
74 Adds a new book to the database based on the provided book form. This function
75 processes the data from the book form, creates a new Book object, and saves
76 it to the database. If any duplicate or invalid data causes issues, it will
77 handle the exceptions accordingly.
79 :param book_form: The form containing information about the book to be added.
81 :type book_form: BookForm
83 :return: The newly added Book object.
85 :rtype: Book
87 :raises ValueError: If a book with the same unique constraint already exists.
88 :raises RuntimeError: If a database request error or any unexpected error occurs.
89 """
90 try:
91 # Create and add the new book
92 new_book = Book(
93 author=sanitize(book_form.author.data),
94 title=sanitize(book_form.title.data),
95 asin=sanitize(book_form.asin.data),
96 link=book_form.link.data, # form validator checked this
97 image=book_form.image.data, # form validator checked this
98 categories_flat=sanitize_categories_flat(book_form.categories_flat.data),
99 book_description=sanitize(book_form.book_description.data),
100 rating=book_form.rating.data or 0.0,
101 isbn_13=sanitize(book_form.isbn_13.data),
102 isbn_10=sanitize(book_form.isbn_10.data),
103 hardcover=sanitize(book_form.hardcover.data),
104 bestsellers_rank_flat=sanitize(book_form.bestsellers_rank_flat.data),
105 specifications_flat=sanitize(book_form.specifications_flat.data),
106 )
107 db.session.add(new_book)
108 db.session.commit()
109 except IntegrityError:
110 db.session.rollback()
111 raise ValueError("A book with the same unique constraint already exists.")
112 except InvalidRequestError as e:
113 db.session.rollback()
114 raise RuntimeError(f"Database request error: {e}")
115 except Exception as e:
116 db.session.rollback()
117 raise RuntimeError(f"An unexpected error occurred: {e}")
118 return new_book
121def update_book(book_form: BookForm) -> Book:
122 """
123 Updates the details of an existing book in the database. It retrieves the book
124 based on the provided ID in the book form, updates the book's attributes with new
125 data, and commits the changes to the database. In the case of any errors during
126 the database operation, the session is rolled back and appropriate exceptions
127 are raised.
129 :param book_form: Form containing updated data for a book
130 :type book_form: BookForm
131 :return: The updated book object
132 :rtype: Book
133 :raises ValueError: Raised when a unique constraint in the database is violated
134 :raises RuntimeError: Raised in case of a database request error or any unexpected error
135 """
136 try:
137 book = get_book_by_id(book_form.id.data)
138 # Update the new book
139 book.author = sanitize(book_form.author.data)
140 book.title = sanitize(book_form.title.data)
141 book.asin = sanitize(book_form.asin.data)
142 book.link = book_form.link.data # handled in the form validator
143 book.image = book_form.image.data # handled in the form validator
144 book.categories_flat = sanitize_categories_flat(book_form.categories_flat.data)
145 book.book_description = sanitize(book_form.book_description.data)
146 book.rating = book_form.rating.data or 0.0
147 book.isbn_13 = sanitize(book_form.isbn_13.data)
148 book.isbn_10 = sanitize(book_form.isbn_10.data)
149 book.hardcover = sanitize(book_form.hardcover.data)
150 book.bestsellers_rank_flat = sanitize(book_form.bestsellers_rank_flat.data)
151 book.specifications_flat = sanitize(book_form.specifications_flat.data)
153 # Update the book in the database
154 db.session.commit()
155 except IntegrityError:
156 db.session.rollback()
157 raise ValueError("A book with the same unique constraint already exists.")
158 except InvalidRequestError as e:
159 db.session.rollback()
160 raise RuntimeError(f"Database request error: {e}")
161 except Exception as e:
162 db.session.rollback()
163 raise RuntimeError(f"An unexpected error occurred: {e}")
164 return book
167def del_book(book_id):
168 """
169 Deletes a book from the database based on the provided book ID.
171 :param book_id: The ID of the book to delete.
172 :type book_id: int
173 :raises ValueError: If no book with the specified ID is found.
174 """
175 try:
176 book = get_book_by_id(book_id)
177 if not book: 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true
178 raise ValueError(f"Book with ID {book_id} not found.")
180 db.session.delete(book)
181 db.session.commit()
182 except Exception as e:
183 db.session.rollback()
184 raise RuntimeError(f"An error occurred while deleting the book: {e}")
187def get_book_status(book_id, user_id) -> ReadingStatusEnum:
188 """
189 Retrieve the reading status of a book for a specific user.
191 This function queries the database for the reading status of a book
192 associated with a specific user. If a status is found, it is returned;
193 otherwise, the function will return None.
195 :param book_id: The unique identifier of the book for which the reading
196 status is being queried.
197 :type book_id: int
198 :param user_id: The unique identifier of the user for whom the reading
199 status is being queried.
200 :type user_id: int
201 :return: The reading status of the book for the user, represented as
202 a value in the ReadingStatusEnum enumeration, or None if no status
203 is found.
204 :rtype: ReadingStatusEnum | None
205 """
206 query = (db.session.query(ReadingStatus.status)
207 .filter(ReadingStatus.book_id == book_id, ReadingStatus.user_id == user_id))
208 row = query.one_or_none()
209 return row.status if row else None
212def get_book_feedback(book_id, user_id) -> FeedbackEnum:
213 """
214 Retrieves feedback for a specified book from a specific user. This function queries the
215 database for a feedback record that matches the provided book ID and user ID. If a
216 matching record exists, the function returns the feedback associated with it. If no
217 record is found, it returns None.
219 :param book_id: ID of the book for which feedback is being retrieved
220 :param user_id: ID of the user providing the feedback
221 :return: FeedbackEnum representing the feedback if found, otherwise None
222 """
223 query = (db.session.query(Feedback.feedback)
224 .filter(Feedback.book_id == book_id, Feedback.user_id == user_id))
225 row = query.one_or_none()
226 return row.feedback if row else None
229def set_book_status(book_id: int, status: str, user_id: int) -> dict:
230 """
231 Set the reading status of a book for a specific user. This function allows the user
232 to update their reading status of a particular book, delete an existing status, or
233 add a new status if it doesn't already exist. If the status is set to "none", any
234 existing status for that book and user combination will be removed.
236 :param book_id: The identifier of the book whose reading status is being modified.
237 :type book_id: int
238 :param status: The new reading status for the book. Accepted values are the status
239 strings such as "none", "reading", "completed", etc.
240 :type status: str
241 :param user_id: The identifier of the user whose book reading status is being updated.
242 :type user_id: int
243 :return: A dictionary containing the updated details of the book, including its
244 status and feedback for the given user.
245 :rtype: dict
246 """
247 # Check if the reading status already exists for this user_id and book_id
248 existing_status = db.session.query(
249 ReadingStatus.id,
250 ReadingStatus.status).filter_by(
251 user_id=user_id,
252 book_id=book_id
253 ).first()
255 # If "none", delete the reading status
256 if status == "none":
257 if existing_status: 257 ↛ 276line 257 didn't jump to line 276 because the condition on line 257 was always true
258 db.session.execute(
259 delete(ReadingStatus)
260 .where(ReadingStatus.id == existing_status.id)
261 )
262 else:
263 if existing_status:
264 # Update the current reading status
265 db.session.execute(
266 update(ReadingStatus)
267 .where(ReadingStatus.id == existing_status.id)
268 .values(status=status)
269 )
270 else:
271 # Add a new reading status
272 new_status = ReadingStatus(user_id=user_id, book_id=book_id, status=status)
273 db.session.add(new_status)
275 # Commit the transaction
276 db.session.commit()
279def set_book_feedback(book_id: int, fb: str, user_id: int) -> dict:
280 """
281 Updates or removes feedback for a specific book and user. If a feedback entry exists for
282 the given user and book, it will be updated based on the provided feedback value. If the
283 feedback is marked as "none", the existing feedback will be removed. If no feedback exists,
284 a new entry will be created. Finally, the updated book details are fetched and returned.
285 """
286 # Check if feedback already exists for this user_id and book_id
287 existing_feedback = db.session.query(
288 Feedback.id,
289 Feedback.feedback).filter_by(
290 user_id=user_id,
291 book_id=book_id
292 ).first()
294 # If "none", delete the feedback
295 if fb == "none":
296 if existing_feedback: 296 ↛ 315line 296 didn't jump to line 315 because the condition on line 296 was always true
297 db.session.execute(
298 delete(Feedback)
299 .where(Feedback.id == existing_feedback.id)
300 )
301 else:
302 if existing_feedback:
303 # Update existing feedback
304 db.session.execute(
305 update(Feedback)
306 .where(Feedback.id == existing_feedback.id)
307 .values(feedback=fb)
308 )
309 else:
310 # Insert new feedback if it wasn't there
311 new_feedback = Feedback(user_id=user_id, book_id=book_id, feedback=fb)
312 db.session.add(new_feedback)
314 # Commit the transaction
315 db.session.commit()
318def book_to_dict_with_status_and_feedback(book, user_id):
319 """
320 Converts a book object to a dictionary representation while appending user's
321 feedback and reading status when available. This function ensures that the
322 resulting dictionary contains accurately mapped data for feedback and status
323 for a specified user.
325 If `book.feedbacks` or `book.reading_statuses` are available, the feedback
326 and status values are extracted from these lists respectively. If unavailable
327 or empty, the function queries for the user's feedback or status using
328 appropriate functions. Default values are assigned when feedback or status
329 is missing or uninitialized.
331 Parameters
332 ----------
333 :param book:
334 The book object containing attributes and possibly associated feedbacks
335 and reading statuses.
336 :param user_id:
337 The unique identifier of the user for whom feedback and status are
338 retrieved.
340 Returns
341 -------
342 :return:
343 A dictionary representation of the book object with additional `feedback`
344 and `status` keys reflecting the user's interaction with the book.
345 :rtype: dict
346 """
347 book_dict = book.to_dict()
348 if user_id: 348 ↛ 372line 348 didn't jump to line 372 because the condition on line 348 was always true
349 if book.feedbacks is not None and book.feedbacks != []:
350 # Assume it's a list with one thing in it, grab the first Feedback and get feedback
351 book_dict['feedback'] = book.feedbacks[0].feedback.value
352 elif book.feedbacks is None: 352 ↛ 354line 352 didn't jump to line 354 because the condition on line 352 was never true
353 # Know nothing, query feedback then
354 fb = get_book_feedback(book_dict['id'], user_id)
355 if fb:
356 book_dict['feedback'] = fb.value
357 else:
358 # Empty list, then, no feedback. Set to None
359 book_dict['feedback'] = 'none'
361 if book.reading_statuses is not None and book.reading_statuses != []:
362 # Assume it's a list with one thing in it, get the first Reading_Status and get status
363 book_dict['status'] = book.reading_statuses[0].status.value
364 elif book.reading_statuses is None: 364 ↛ 366line 364 didn't jump to line 366 because the condition on line 364 was never true
365 # Know noting, query db
366 status = get_book_status(book_dict['id'], user_id)
367 if status:
368 book_dict['status'] = status.value
369 else:
370 # Empty list, then no status. Set to None
371 book_dict['status'] = 'none'
372 return book_dict