First of three PRs that convert Overdrive API responses to pydantic models (PP-3175)

Author: dbernstein

src/palace/manager/integration/license/overdrive/api.py

@@ -78,6 +78,7 @@
     OverdriveManifestFulfillment,
 )
 from palace.manager.integration.license.overdrive.model import (
+    Availability,
     BaseOverdriveModel,
     Checkout,
     Checkouts,
@@ -1692,25 +1693,31 @@ def update_licensepool(
                 status_code,
             )
             return None, None, False
-        book.update(json.loads(content))
 
-        # Update book_id now that we know we have new data.
-        book_id = book["id"]
+        availability = Availability.model_validate_json(content)
+
+        # Use the caller-provided ID for LicensePool lookup. This is always
+        # set because circulation_lookup creates dict(id=book_id) when called
+        # with a string, and book-list dicts always include "id".
+        resolved_book_id = cast(str, book.get("id"))
+
         license_pool, is_new = LicensePool.for_foreign_id(
             self._db,
             DataSource.OVERDRIVE,
             Identifier.OVERDRIVE_ID,
-            cast(str, book_id),
+            resolved_book_id,
             collection=self.collection,
         )
         if is_new or not license_pool.work:
-            # Either this is the first time we've seen this book or its doesn't
+            # Either this is the first time we've seen this book or it doesn't
             # have an associated work. Make sure its identifier has bibliographic coverage.
             self.overdrive_bibliographic_coverage_provider.ensure_coverage(
                 license_pool.identifier, force=True
             )
 
-        return self.update_licensepool_with_book_info(book, license_pool, is_new)
+        return self.update_licensepool_with_book_info(
+            availability, resolved_book_id, license_pool, is_new
+        )
 
     # Alias for the CirculationAPI interface
     def update_availability(self, licensepool: LicensePool) -> None:
@@ -1728,18 +1735,26 @@ def _edition(self, licensepool: LicensePool) -> tuple[Edition, bool]:
         )
 
     def update_licensepool_with_book_info(
-        self, book: dict[str, Any], license_pool: LicensePool, is_new_pool: bool
+        self,
+        availability: Availability,
+        book_id: str,
+        license_pool: LicensePool,
+        is_new_pool: bool,
     ) -> tuple[LicensePool, bool, bool]:
-        """Update a book's LicensePool with information from a JSON
-        representation of its circulation info.
+        """Update a book's LicensePool with information from an availability document.
 
         Then, create an Edition and make sure it has bibliographic
         coverage. If the new Edition is the only candidate for the
         pool's presentation_edition, promote it to presentation
         status.
+
+        :param availability: The parsed Overdrive availability document.
+        :param book_id: The Overdrive product ID for this book.
+        :param license_pool: The LicensePool to update.
+        :param is_new_pool: Whether this is a newly created LicensePool.
         """
         extractor = OverdriveRepresentationExtractor(self)
-        circulation = extractor.book_info_to_circulation(book)
+        circulation = extractor.book_info_to_circulation(availability, book_id=book_id)
         lp, circulation_changed = circulation.apply(self._db, license_pool.collection)
         if lp is not None:
             license_pool = lp

src/palace/manager/integration/license/overdrive/importer.py

@@ -18,6 +18,7 @@
     BookInfoEndpoint,
     OverdriveAPI,
 )
+from palace.manager.integration.license.overdrive.model import Availability
 from palace.manager.integration.license.overdrive.representation import (
     OverdriveRepresentationExtractor,
 )
@@ -162,15 +163,16 @@ def _process_book(
 
         # availability needs to be checked/updated in all but a few instances so it is
         # probably not worth the compute time to save ourselves a handful of unnecessary updates.
-        availability = book.get("availabilityV2", None)
-        if not availability:
+        availability_data = book.get("availabilityV2", None)
+        if not availability_data:
             # This is a rare and probably transient case where the availabilityV2
             # was not retrieved due to a 404 from OD.
             self.log.warning(
                 f"No availabilityV2 found for book {identifier}. book={book}.  This state can "
                 f"arise when the OD returns a 404 for the availability url."
             )
         else:
+            availability = Availability.model_validate(availability_data)
             circulation = self._extractor.book_info_to_circulation(availability)
             # The circulation should never be null here because there is a non-null entry for availabilityV2 in the
             # book dictionary.  Mypy complains without an assertion or type hints.

src/palace/manager/integration/license/overdrive/model.py

@@ -1,6 +1,7 @@
 import json
 import re
 import typing
+from enum import StrEnum
 from functools import cached_property
 from typing import Protocol, Self, overload
 from urllib.parse import quote_plus
@@ -494,3 +495,49 @@ class PatronInformation(BaseOverdriveModel):
         default_factory=dict, alias="linkTemplates"
     )
     actions: list[dict[str, Action]] = Field(default_factory=list)
+
+
+class AvailabilityType(StrEnum):
+    """Availability type for a title in an Overdrive collection."""
+
+    NORMAL = "Normal"
+    ALWAYS_AVAILABLE = "AlwaysAvailable"
+    LIMITED_AVAILABILITY = "LimitedAvailability"
+
+
+class AvailabilityAccount(BaseOverdriveModel):
+    """
+    Per-library copy availability within an Overdrive availability document.
+
+    See: https://developer.overdrive.com/apis/library-availability-new
+    """
+
+    id: int
+    copies_owned: NonNegativeInt = Field(0, alias="copiesOwned")
+    copies_available: NonNegativeInt = Field(0, alias="copiesAvailable")
+    shared: bool = False
+
+
+class Availability(BaseOverdriveModel):
+    """
+    Availability information for a single title.
+
+    This model is used for both successful availability responses and error
+    responses (e.g. ``errorCode: "NotFound"``). Because error responses do not
+    include ``reserveId``, that field is optional. Callers that receive an error
+    response must supply the book ID from context via the ``book_id`` parameter
+    of :meth:`OverdriveRepresentationExtractor.book_info_to_circulation`.
+
+    See: https://developer.overdrive.com/apis/library-availability-new
+    """
+
+    reserve_id: str | None = Field(None, alias="reserveId")
+    accounts: list[AvailabilityAccount] = Field(default_factory=list)
+    availability_type: AvailabilityType | None = Field(None, alias="availabilityType")
+    available: bool = False
+    copies_available: NonNegativeInt | None = Field(None, alias="copiesAvailable")
+    copies_owned: NonNegativeInt | None = Field(None, alias="copiesOwned")
+    number_of_holds: NonNegativeInt = Field(0, alias="numberOfHolds")
+    is_owned_by_collections: bool | None = Field(None, alias="isOwnedByCollections")
+    error_code: str | None = Field(None, alias="errorCode")
+    links: dict[str, Link] = Field(default_factory=dict)

src/palace/manager/integration/license/overdrive/representation.py

@@ -18,6 +18,10 @@
 from palace.manager.integration.license.overdrive.constants import (
     OVERDRIVE_MAIN_ACCOUNT_ID,
 )
+from palace.manager.integration.license.overdrive.model import (
+    Availability,
+    AvailabilityAccount,
+)
 from palace.manager.integration.license.overdrive.util import _make_link_safe
 from palace.manager.sqlalchemy.constants import MediaTypes
 from palace.manager.sqlalchemy.model.classification import Classification, Subject
@@ -219,31 +223,41 @@ def parse_roles(cls, id: str, rolestring: str) -> list[Contributor.Role]:
                 processed.append(cls.overdrive_role_to_simplified_role[x])
         return processed
 
-    def book_info_to_circulation(self, book: dict[str, Any]) -> CirculationData:
-        """Note:  The json data passed into this method is from a different file/stream
-        from the json data that goes into the book_info_to_metadata() method.
+    def book_info_to_circulation(
+        self, availability: Availability, book_id: str | None = None
+    ) -> CirculationData:
+        """Convert an Overdrive availability document into a :class:`CirculationData` object.
+
+        Note: The availability data passed into this method is from a different
+        API endpoint than the metadata data that goes into
+        :meth:`book_info_to_bibliographic`.
+
+        :param availability: The parsed Overdrive availability document.
+        :param book_id: Optional Overdrive ID to use when the availability
+            document does not include a ``reserveId`` (e.g. a NotFound error
+            response).  If neither ``availability.reserve_id`` nor ``book_id``
+            is present, a :exc:`PalaceValueError` is raised.
         """
-        # In Overdrive, 'reserved' books show up as books on
-        # hold. There is no separate notion of reserved books.
+        # In Overdrive, 'reserved' books show up as books on hold.
         licenses_reserved = 0
 
         licenses_owned = None
         licenses_available = None
         patrons_in_hold_queue = None
 
-        # TODO: The only reason this works for a NotFound error is the
-        # circulation code sticks the known book ID into `book` ahead
-        # of time. That's a code smell indicating that this system
-        # needs to be refactored.
-        if "reserveId" in book and not "id" in book:
-            book["id"] = book["reserveId"]
-        if not "id" in book:
-            self.log.error("Book has no ID: %r", book)
+        # book_id takes precedence over the document's reserveId so that the
+        # caller can override the identifier (e.g. after a circulation lookup
+        # where the caller already knows the book's ID). Fall back to reserveId
+        # when no external book_id is provided (e.g. from the importer).
+        overdrive_id = book_id or availability.reserve_id
+        if not overdrive_id:
+            self.log.error("Availability has no ID: %r", availability)
             raise PalaceValueError("Book must have an id to be processed")
-        overdrive_id = book["id"]
+
         primary_identifier = IdentifierData(
             type=Identifier.OVERDRIVE_ID, identifier=overdrive_id
         )
+
         # TODO: We might be able to use this information to avoid the
         # need for explicit configuration of Advantage collections, or
         # at least to keep Advantage collections more up-to-date than
@@ -261,28 +275,24 @@ def book_info_to_circulation(self, book: dict[str, Any]) -> CirculationData:
         # similarly, though those can abruptly become unavailable, so
         # UNLIMITED_ACCESS is probably not appropriate.
 
-        error_code = book.get("errorCode")
         # TODO: It's not clear what other error codes there might be.
         # The current behavior will respond to errors other than
         # NotFound by leaving the book alone, but this might not be
         # the right behavior.
-        if error_code == "NotFound":
+        if availability.error_code == "NotFound":
             licenses_owned = 0
             licenses_available = 0
             patrons_in_hold_queue = 0
-        elif book.get("isOwnedByCollections") is not False:
+        elif availability.is_owned_by_collections is not False:
             # We own this book.
             licenses_owned = 0
             licenses_available = 0
 
-            for account in self._get_applicable_accounts(book.get("accounts", [])):
-                licenses_owned += int(account.get("copiesOwned", 0))
-                licenses_available += int(account.get("copiesAvailable", 0))
+            for account in self._get_applicable_accounts(availability.accounts):
+                licenses_owned += account.copies_owned
+                licenses_available += account.copies_available
 
-            if "numberOfHolds" in book:
-                if patrons_in_hold_queue is None:
-                    patrons_in_hold_queue = 0
-                patrons_in_hold_queue += book["numberOfHolds"]
+            patrons_in_hold_queue = availability.number_of_holds
 
         if licenses_owned is None:
             license_status = None
@@ -302,38 +312,26 @@ def book_info_to_circulation(self, book: dict[str, Any]) -> CirculationData:
         )
 
     def _get_applicable_accounts(
-        self, accounts: list[dict[str, Any]]
-    ) -> list[dict[str, Any]]:
-        """
-        Returns those accounts from the accounts array that apply the
-        current overdrive collection context.
+        self, accounts: list[AvailabilityAccount]
+    ) -> list[AvailabilityAccount]:
+        """Return the accounts from the availability document that apply to the
+        current Overdrive collection context.
 
-        If this is an overdrive parent collection, we want to return accounts
-        associated with the main OverDrive "library" and any non-main account
-        with sharing enabled.
+        For a parent collection, returns accounts for the main OverDrive
+        "library" and any sub-account with sharing enabled.
 
-        If this is a child OverDrive collection, then we return only the
-        account associated with that child's OverDrive Advantage "library".
-        Additionally, we want to exclude the account if it is "shared" since
-        we will be counting it with the parent collection.
+        For a child Overdrive Advantage collection, returns only the account
+        matching that child's library ID (excluding shared copies, which are
+        counted with the parent).
         """
-
         if self.library_id == OVERDRIVE_MAIN_ACCOUNT_ID:
-            # this is a parent collection
-            filtered_result = filter(
-                lambda account: account.get("id") == OVERDRIVE_MAIN_ACCOUNT_ID
-                or account.get("shared", False),
-                accounts,
-            )
+            # parent collection
+            return [
+                a for a in accounts if a.id == OVERDRIVE_MAIN_ACCOUNT_ID or a.shared
+            ]
         else:
-            # this is child collection
-            filtered_result = filter(
-                lambda account: account.get("id") == self.library_id
-                and not account.get("shared", False),
-                accounts,
-            )
-
-        return list(filtered_result)
+            # child Advantage collection
+            return [a for a in accounts if a.id == self.library_id and not a.shared]
 
     @classmethod
     def image_link_to_linkdata(