PP-129 OPDS 2 with acquisition links

api/controller.py

@@ -21,6 +21,7 @@
 from sqlalchemy import select
 from sqlalchemy.orm import eagerload
 from sqlalchemy.orm.exc import NoResultFound
+from werkzeug.datastructures import MIMEAccept
 
 from api.authentication.access_token import AccessTokenProvider
 from api.model.patron_auth import PatronAuthAccessToken
@@ -1323,7 +1324,7 @@ def publications(self):
             self.search_engine,
         )
         return feed.as_response(
-            mime_types=[("application/opds+json", 1)],  # Force the type
+            mime_types=MIMEAccept([("application/opds+json", 1)]),  # Force the type
             max_age=int(max_age) if max_age is not None else None,
         )
 

core/feed_protocol/navigation.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, List, Optional, Tuple
+from typing import Any, Optional
 
 from sqlalchemy.orm import Session
 from typing_extensions import Self
@@ -84,7 +84,7 @@ def add_entry(
 
     def as_response(
         self,
-        mime_types: Optional[MIMEAccept | List[Tuple[str, float]]] = None,
+        mime_types: Optional[MIMEAccept] = None,
         **kwargs: Any,
     ) -> OPDSFeedResponse:
         response = super().as_response(mime_types=mime_types, **kwargs)

core/feed_protocol/opds.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Type
 
 from werkzeug.datastructures import MIMEAccept
 
@@ -14,17 +14,19 @@
 
 
 def get_serializer(
-    mime_types: Optional[MIMEAccept | List[Tuple[str, float]]],
+    mime_types: Optional[MIMEAccept],
 ) -> OPDS1Serializer | OPDS2Serializer:
     # Loop through and return whichever mimetype is encountered first
     # Sort values by q-value first
-    if mime_types is not None:
-        mime_types = sorted(mime_types, key=lambda mime: mime[1], reverse=True)
-    for mime in mime_types or []:
-        if "application/opds+json" in mime[0]:
-            return OPDS2Serializer()
-        elif "application/atom+xml" in mime[0]:
-            return OPDS1Serializer()
+    serializers: Dict[str, Type[Any]] = {
+        "application/opds+json": OPDS2Serializer,
+        "application/atom+xml": OPDS1Serializer,
+    }
+    if mime_types:
+        match = mime_types.best_match(
+            serializers.keys(), default="application/atom+xml"
+        )
+        return serializers[match]()  # type: ignore[no-any-return]
     # Default
     return OPDS1Serializer()
 
@@ -42,9 +44,7 @@ def __init__(
         self._feed = FeedData()
         self.log = logging.getLogger(self.__class__.__name__)
 
-    def serialize(
-        self, mime_types: Optional[MIMEAccept | List[Tuple[str, float]]] = None
-    ) -> bytes:
+    def serialize(self, mime_types: Optional[MIMEAccept] = None) -> bytes:
         serializer = get_serializer(mime_types)
         return serializer.serialize_feed(self._feed)
 
@@ -53,7 +53,7 @@ def add_link(self, href: str, rel: Optional[str] = None, **kwargs: Any) -> None:
 
     def as_response(
         self,
-        mime_types: Optional[MIMEAccept | List[Tuple[str, float]]] = None,
+        mime_types: Optional[MIMEAccept] = None,
         **kwargs: Any,
     ) -> OPDSFeedResponse:
         """Serialize the feed using the serializer protocol"""
@@ -70,7 +70,7 @@ def as_response(
     def entry_as_response(
         cls,
         entry: WorkEntry,
-        mime_types: Optional[MIMEAccept | List[Tuple[str, float]]] = None,
+        mime_types: Optional[MIMEAccept] = None,
         **response_kwargs: Any,
     ) -> OPDSEntryResponse:
         if not entry.computed:

tests/api/feed_protocol/test_opds_acquisition_feed.py

@@ -6,6 +6,7 @@
 
 import pytest
 from sqlalchemy.orm import Session
+from werkzeug.datastructures import MIMEAccept
 
 from core.entrypoint import (
     AudiobooksEntryPoint,
@@ -63,7 +64,7 @@ def test_entry_as_response(self, db: DatabaseTransactionFixture):
 
         # Specifically asking for a json type
         response = BaseOPDSFeed.entry_as_response(
-            entry, mime_types=[("application/opds+json", 0.9)]
+            entry, mime_types=MIMEAccept([("application/opds+json", 0.9)])
         )
         assert isinstance(response, OPDSEntryResponse)
         assert response.content_type == "application/opds+json"

tests/api/feed_protocol/test_opds_base.py

@@ -30,13 +30,13 @@ def test_get_serializer(self):
         )
         assert isinstance(get_serializer(request.accept_mimetypes), OPDS1Serializer)
 
-        # The default q-value should be 1
+        # The default q-value should be 1, but opds2 specificity is higher
         request = Request.from_values(
             headers=dict(
                 Accept="application/atom+xml;profile=feed,application/opds+json;q=0.9"
             )
         )
-        assert isinstance(get_serializer(request.accept_mimetypes), OPDS1Serializer)
+        assert isinstance(get_serializer(request.accept_mimetypes), OPDS2Serializer)
 
         # The default q-value should sort above 0.9
         request = Request.from_values(
@@ -55,11 +55,3 @@ def test_get_serializer(self):
         # No valid accept mimetype should default to OPDS1.x
         request = Request.from_values(headers=dict(Accept="text/html"))
         assert isinstance(get_serializer(request.accept_mimetypes), OPDS1Serializer)
-
-        # Sorting occurs on list type values
-        assert isinstance(
-            get_serializer(
-                [("application/opds+json", 0.9), ("application/atom+xml", 1)]
-            ),
-            OPDS1Serializer,
-        )