From 5522bcea79542cb8773afe5f8c473f2160632e65 Mon Sep 17 00:00:00 2001 From: Waiming Yuen Date: Thu, 11 Jun 2026 11:39:19 -0700 Subject: [PATCH] Improve samples/book-app-project/books.py: add atomic save, validation, context manager, comparison ops Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- samples/book-app-project/books.py | 203 ++++++++++++++++++++++++++---- 1 file changed, 182 insertions(+), 21 deletions(-) diff --git a/samples/book-app-project/books.py b/samples/book-app-project/books.py index 2110689f..8ef6b49c 100644 --- a/samples/book-app-project/books.py +++ b/samples/book-app-project/books.py @@ -1,8 +1,23 @@ +"""Utilities for Book and BookCollection used in samples/book-app-project. + +Provides a small API for managing a persistent collection of books stored as JSON. +""" + import json from dataclasses import dataclass, asdict -from typing import List, Optional +from typing import List, Optional, Dict, Any, Union +from pathlib import Path +import tempfile +import os +import logging +from datetime import datetime + +__all__ = ["Book", "BookCollection"] -DATA_FILE = "data.json" +LOG = logging.getLogger(__name__) +LOG.addHandler(logging.NullHandler()) + +DATA_FILE = Path(__file__).parent / "data.json" @dataclass @@ -12,47 +27,156 @@ class Book: year: int read: bool = False + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "Book": + return cls( + title=str(data["title"]), + author=str(data["author"]), + year=int(data["year"]), + read=bool(data.get("read", False)), + ) + + def __repr__(self) -> str: # helpful for REPL and tests + return f"Book(title={self.title!r}, author={self.author!r}, year={self.year}, read={self.read})" + + def __eq__(self, other: object) -> bool: + if not isinstance(other, Book): + return NotImplemented + return (self.title.lower(), self.author.lower(), self.year) == ( + other.title.lower(), other.author.lower(), other.year + ) + + def __lt__(self, other: "Book") -> bool: + return (self.title.lower(), self.author.lower(), self.year) < ( + other.title.lower(), other.author.lower(), other.year + ) + + +def _validate_year(year: int) -> int: + now_year = datetime.utcnow().year + y = int(year) + if y < 0 or y > now_year + 1: + raise ValueError(f"year {y} is not reasonable") + return y + class BookCollection: - def __init__(self): + """A simple collection of Book objects with JSON persistence. + + Data file defaults to samples/book-app-project/data.json (next to this file), + but a custom path may be passed to the constructor (useful for tests). + """ + + def __init__(self, data_file: Optional[Path] = None): + self.data_file: Path = Path(data_file) if data_file else DATA_FILE self.books: List[Book] = [] self.load_books() - def load_books(self): - """Load books from the JSON file if it exists.""" + def __enter__(self) -> "BookCollection": + return self + + def __exit__(self, exc_type, exc, tb) -> bool: + # On normal exit persist changes; never suppress exceptions. try: - with open(DATA_FILE, "r") as f: - data = json.load(f) - self.books = [Book(**b) for b in data] - except FileNotFoundError: + self.save_books() + except Exception: + LOG.exception("Failed to save books on __exit__") + return False + + def load_books(self) -> None: + """Load books from the JSON file if it exists. + + If the file is corrupt, it is moved to a timestamped backup so the user can inspect it. + """ + if not self.data_file.exists(): self.books = [] + return + try: + with self.data_file.open("r", encoding="utf-8") as f: + data = json.load(f) + self.books = [Book.from_dict(b) for b in data] except json.JSONDecodeError: - print("Warning: data.json is corrupted. Starting with empty collection.") + # Back up corrupted file for inspection and start with empty collection + ts = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ") + corrupt_name = self.data_file.with_suffix(f".corrupt.{ts}.json") + try: + self.data_file.replace(corrupt_name) + LOG.warning("Moved corrupted %s to %s", self.data_file, corrupt_name) + except Exception: + LOG.exception("Failed to back up corrupted data file %s", self.data_file) self.books = [] - def save_books(self): - """Save the current book collection to JSON.""" - with open(DATA_FILE, "w") as f: - json.dump([asdict(b) for b in self.books], f, indent=2) + def save_books(self) -> None: + """Atomically save the current book collection to JSON. + + Ensures parent directory exists and writes via a temp file then os.replace. + """ + self.data_file.parent.mkdir(parents=True, exist_ok=True) + data = [b.to_dict() for b in self.books] + # Write to a temp file first, then atomically replace to avoid corruption. + tmp_fd, tmp_path = tempfile.mkstemp(dir=str(self.data_file.parent)) + try: + with os.fdopen(tmp_fd, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, ensure_ascii=False, sort_keys=True) + os.replace(tmp_path, str(self.data_file)) + except Exception: + # Attempt to clean up temp file on failure + try: + os.remove(tmp_path) + except OSError: + pass + raise def add_book(self, title: str, author: str, year: int) -> Book: - book = Book(title=title, author=author, year=year) + """Add a book if not already present (title+author considered unique). + + Raises ValueError if the book already exists or year is invalid. + """ + title_clean = title.strip() + author_clean = author.strip() + y = _validate_year(year) + if self.find_book_by_title_and_author(title_clean, author_clean): + raise ValueError("Book already exists") + book = Book(title=title_clean, author=author_clean, year=y) self.books.append(book) self.save_books() return book - def list_books(self) -> List[Book]: - return self.books + def list_books(self, sort_by: Optional[str] = None) -> List[Book]: + """Return list of books; optionally sort by 'title', 'author', or 'year'.""" + if sort_by in ("title", "author", "year"): + return sorted(self.books, key=lambda b: getattr(b, sort_by)) + return list(self.books) + + def search_title(self, query: str) -> List[Book]: + """Return books whose title contains the query (case-insensitive).""" + q = query.strip().lower() + return [b for b in self.books if q in b.title.lower()] def find_book_by_title(self, title: str) -> Optional[Book]: + """Find first exact-title match (case-insensitive).""" + t = title.strip().lower() + matches = [book for book in self.books if book.title.lower() == t] + if not matches: + return None + if len(matches) > 1: + LOG.warning("Multiple books found with title %s; returning first. Consider using title+author.", title) + return matches[0] + + def find_book_by_title_and_author(self, title: str, author: str) -> Optional[Book]: + t = title.strip().lower() + a = author.strip().lower() for book in self.books: - if book.title.lower() == title.lower(): + if book.title.lower() == t and book.author.lower() == a: return book return None def mark_as_read(self, title: str) -> bool: book = self.find_book_by_title(title) - if book: + if book and not book.read: book.read = True self.save_books() return True @@ -68,5 +192,42 @@ def remove_book(self, title: str) -> bool: return False def find_by_author(self, author: str) -> List[Book]: - """Find all books by a given author.""" - return [b for b in self.books if b.author.lower() == author.lower()] + """Find all books where the provided string is contained in the author name (case-insensitive).""" + a = author.strip().lower() + return [b for b in self.books if a in b.author.lower()] + + def update_book(self, title: str, **fields) -> bool: + """Update fields on a book identified by title. Allowed fields: title, author, year, read. + + Returns True if the book was found and updated; False otherwise. + """ + book = self.find_book_by_title(title) + if not book: + return False + allowed = {"title", "author", "year", "read"} + changed = False + for k, v in fields.items(): + if k in allowed: + if k == "year": + v = _validate_year(v) + setattr(book, k, v) + changed = True + if changed: + self.save_books() + return changed + + def clear_books(self) -> None: + """Remove all books from the collection and persist the empty state.""" + self.books = [] + self.save_books() + + def count(self) -> int: + """Return number of books in the collection.""" + return len(self.books) + + def export_json(self, path: Path) -> None: + """Export the collection to the given path as JSON.""" + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as f: + json.dump([b.to_dict() for b in self.books], f, indent=2, ensure_ascii=False, sort_keys=True)