Add Pydantic models for RWPM, OPDS2, and ODL

pyproject.toml

@@ -244,6 +244,7 @@ opensearch-dsl = "~1.0"
 opensearch-py = "~1.1"
 palace-webpub-manifest-parser = "^4.0.0"
 pillow = "^11.0"
+pycountry = "^24.6.1"
 pycryptodome = "^3.18"
 pydantic = {version = "^2.9.2", extras = ["email"]}
 pydantic-settings = "^2.5.2"

src/palace/manager/api/odl/api.py

@@ -51,9 +51,9 @@
 )
 from palace.manager.core.exceptions import PalaceValueError
 from palace.manager.core.lcp.credential import LCPCredentialFactory
-from palace.manager.opds.base import BaseLink
 from palace.manager.opds.lcp.license import LicenseDocument
 from palace.manager.opds.lcp.status import LoanStatus
+from palace.manager.opds.types.link import BaseLink
 from palace.manager.service.container import Services
 from palace.manager.sqlalchemy.model.collection import Collection
 from palace.manager.sqlalchemy.model.datasource import DataSource

src/palace/manager/opds/authentication.py

@@ -6,8 +6,8 @@
 from pydantic import BaseModel, Field, field_validator
 
 from palace.manager.core.exceptions import PalaceValueError
-from palace.manager.opds.base import ListOfLinks
-from palace.manager.opds.opds import Link
+from palace.manager.opds.rwpm import Link
+from palace.manager.opds.types.link import CompactCollection
 
 
 class AuthenticationLabels(BaseModel):
@@ -18,7 +18,7 @@ class AuthenticationLabels(BaseModel):
 class Authentication(BaseModel):
     type: str
     labels: AuthenticationLabels | None = None
-    links: ListOfLinks[Link]
+    links: CompactCollection[Link]
 
 
 class AuthenticationDocument(BaseModel):
@@ -37,7 +37,7 @@ def content_type(cls) -> str:
     title: str
     authentication: list[Authentication]
     description: str | None = None
-    links: ListOfLinks[Link] = Field(default_factory=ListOfLinks)
+    links: CompactCollection[Link] = Field(default_factory=CompactCollection)
 
     @field_validator("authentication")
     @classmethod

src/palace/manager/opds/base.py

@@ -1,27 +1,6 @@
 from __future__ import annotations
 
-import typing
-from functools import cached_property
-from typing import Any, TypeVar
-
-from pydantic import BaseModel, ConfigDict, GetCoreSchemaHandler
-from pydantic_core import core_schema
-from uritemplate import URITemplate, variable
-
-from palace.manager.core.exceptions import PalaceValueError
-
-T = TypeVar("T")
-
-
-def obj_or_set_to_set(value: T | set[T] | frozenset[T] | None) -> frozenset[T]:
-    """Convert object or set of objects to a set of objects."""
-    if value is None:
-        return frozenset()
-    if isinstance(value, set):
-        return frozenset(value)
-    elif isinstance(value, frozenset):
-        return value
-    return frozenset({value})
+from pydantic import BaseModel, ConfigDict
 
 
 class BaseOpdsModel(BaseModel):
@@ -31,120 +10,3 @@ class BaseOpdsModel(BaseModel):
         populate_by_name=True,
         frozen=True,
     )
-
-
-class BaseLink(BaseOpdsModel):
-    """The various models all have links with this same basic structure, but
-    with additional fields, so we define this base class to avoid repeating
-    the same fields in each model, and so we can use the same basic validation
-    for them all.
-    """
-
-    href: str
-    rel: set[str] | str
-    templated: bool = False
-    type: str | None = None
-
-    @cached_property
-    def rels(self) -> frozenset[str]:
-        return obj_or_set_to_set(self.rel)
-
-    def href_templated(self, var_dict: variable.VariableValueDict | None = None) -> str:
-        """
-        Return the URL with template variables expanded, if necessary.
-        """
-        if not self.templated:
-            return self.href
-        template = URITemplate(self.href)
-        return template.expand(var_dict)
-
-
-LinkT = TypeVar("LinkT", bound="BaseLink")
-
-
-class ListOfLinks(list[LinkT]):
-    """
-    A generic list container for OPDS type links.
-
-    Provides helper methods for finding a link by relation and type, and
-    provides validation to ensure that each href, rel, and type combination
-    in the list is unique.
-    """
-
-    @classmethod
-    def __get_pydantic_core_schema__(
-        cls, source_type: type[Any], handler: GetCoreSchemaHandler
-    ) -> core_schema.CoreSchema:
-        origin_type = typing.get_origin(source_type)
-        assert origin_type is ListOfLinks
-        [container_type] = typing.get_args(source_type)
-        return core_schema.no_info_after_validator_function(
-            cls._validate,
-            core_schema.list_schema(handler(container_type)),
-        )
-
-    @classmethod
-    def _validate(cls, value: list[LinkT]) -> ListOfLinks[LinkT]:
-        link_set = set()
-        links: ListOfLinks[LinkT] = ListOfLinks()
-        for link in value:
-            if (link.rels, link.href, link.type) in link_set:
-                raise PalaceValueError(
-                    f"Duplicate link with relation '{link.rel}', type '{link.type}' and href '{link.href}'"
-                )
-            link_set.add((link.rels, link.href, link.type))
-            links.append(link)
-        return links
-
-    @typing.overload
-    def get(
-        self,
-        *,
-        rel: str | None = ...,
-        type: str | None = ...,
-        raising: typing.Literal[True],
-    ) -> LinkT: ...
-
-    @typing.overload
-    def get(
-        self, *, rel: str | None = ..., type: str | None = ..., raising: bool = ...
-    ) -> LinkT | None: ...
-
-    def get(
-        self, *, rel: str | None = None, type: str | None = None, raising: bool = False
-    ) -> LinkT | None:
-        """
-        Return the link with the specific relation and type. Raises an
-        exception if there are multiple links with the same relation and type.
-        """
-        links = self.get_list(rel=rel, type=type)
-        if (num_links := len(links)) != 1 and raising:
-            if num_links == 0:
-                err = "No links found"
-            else:
-                err = "Multiple links found"
-
-            match (rel, type):
-                case (None, None):
-                    # Nothing to add to the error message
-                    ...
-                case (_, None):
-                    err += f" with rel='{rel}'"
-                case (None, _):
-                    err += f" with type='{type}'"
-                case _:
-                    err += f" with rel='{rel}' and type='{type}'"
-            raise PalaceValueError(err)
-        return next(iter(links), None)
-
-    def get_list(
-        self, *, rel: str | None = None, type: str | None = None
-    ) -> list[LinkT]:
-        """
-        Return links with the specific relation and type.
-        """
-        return [
-            link
-            for link in self
-            if (rel is None or rel in link.rels) and (type is None or type == link.type)
-        ]

src/palace/manager/opds/lcp/license.py

@@ -8,7 +8,8 @@
 )
 
 from palace.manager.core.exceptions import PalaceValueError
-from palace.manager.opds.base import BaseLink, BaseOpdsModel, ListOfLinks
+from palace.manager.opds.base import BaseOpdsModel
+from palace.manager.opds.types.link import BaseLink, CompactCollection
 
 
 class Link(BaseLink):
@@ -106,13 +107,13 @@ def content_type() -> str:
     provider: str
     updated: AwareDatetime | None = None
     encryption: Encryption
-    links: ListOfLinks[Link]
+    links: CompactCollection[Link]
     rights: Rights | None = None
     signature: Signature
 
     @field_validator("links")
     @classmethod
-    def _validate_links(cls, value: ListOfLinks[Link]) -> ListOfLinks[Link]:
+    def _validate_links(cls, value: CompactCollection[Link]) -> CompactCollection[Link]:
         if value.get(rel="hint") is None:
             raise PalaceValueError("links must contain a link with rel 'hint'")
         if value.get(rel="publication") is None:

src/palace/manager/opds/lcp/status.py

@@ -4,7 +4,8 @@
 
 from pydantic import AwareDatetime, Field
 
-from palace.manager.opds.base import BaseLink, BaseOpdsModel, ListOfLinks
+from palace.manager.opds.base import BaseOpdsModel
+from palace.manager.opds.types.link import BaseLink, CompactCollection
 
 # TODO: Remove this when we drop support for Python 3.10
 if sys.version_info >= (3, 11):
@@ -104,7 +105,7 @@ def content_type() -> str:
     status: Status
     message: str
     updated: Updated
-    links: ListOfLinks[Link]
+    links: CompactCollection[Link]
     potential_rights: PotentialRights = Field(default_factory=PotentialRights)
     events: list[Event] = Field(default_factory=list)
 

src/palace/manager/opds/odl/info.py

@@ -1,28 +1,14 @@
-import sys
-from enum import auto
+from collections.abc import Sequence
 from functools import cached_property
 
 from pydantic import AwareDatetime, Field, NonNegativeInt
 
-from palace.manager.opds.base import BaseOpdsModel, obj_or_set_to_set
-from palace.manager.opds.odl.odl import Protection, Terms
-from palace.manager.opds.opds import Price
-
-# TODO: Remove this when we drop support for Python 3.10
-if sys.version_info >= (3, 11):
-    from enum import StrEnum
-else:
-    from backports.strenum import StrEnum
-
-
-class Status(StrEnum):
-    """
-    https://drafts.opds.io/odl-1.0.html#41-syntax
-    """
-
-    PREORDER = auto()
-    AVAILABLE = auto()
-    UNAVAILABLE = auto()
+from palace.manager.opds.base import BaseOpdsModel
+from palace.manager.opds.odl.protection import Protection
+from palace.manager.opds.odl.terms import Terms
+from palace.manager.opds.opds2 import Price
+from palace.manager.opds.util import StrOrTuple, obj_or_tuple_to_tuple
+from palace.manager.sqlalchemy.model.licensing import LicenseStatus
 
 
 class Loan(BaseOpdsModel):
@@ -60,19 +46,20 @@ def content_type() -> str:
         return "application/vnd.odl.info+json"
 
     identifier: str
-    status: Status
+    status: LicenseStatus
     checkouts: Checkouts
-    format: frozenset[str] | str
+    format: StrOrTuple[str] = tuple()
+
+    @cached_property
+    def formats(self) -> Sequence[str]:
+        return obj_or_tuple_to_tuple(self.format)
+
     created: AwareDatetime | None = None
     terms: Terms = Field(default_factory=Terms)
     protection: Protection = Field(default_factory=Protection)
     price: Price | None = None
     source: str | None = None
 
-    @cached_property
-    def formats(self) -> frozenset[str]:
-        return obj_or_set_to_set(self.format)
-
     @cached_property
     def active(self) -> bool:
-        return self.status == Status.AVAILABLE
+        return self.status == LicenseStatus.available