Add pagination support

calibre_rest/calibre.py

@@ -46,6 +46,29 @@ class CalibreWrapper:
         "timestamp",
         "title",
     ]
+    SORT_BY_KEYS = [
+        "author_sort",
+        "authors",
+        "comments",
+        "cover",
+        "formats",
+        "id",
+        "identifiers",
+        "isbn",
+        "languages",
+        "last_modified",
+        "pubdate",
+        "publisher",
+        "rating",
+        "series",
+        "series_index",
+        "size",
+        "tags",
+        "template",
+        "timestamp",
+        "title",
+        "uuid",
+    ]
     ALLOWED_FILE_EXTENSIONS = (
         ".azw",
         ".azw3",
@@ -238,32 +261,88 @@ def get_book(self, id: int) -> Book:
         if len(b) == 1:
             return Book(**b[0])
 
-    # TODO sort and filter
-    def get_books(self, limit: int = 10) -> list[Book]:
+    def get_books(
+        self,
+        limit: int = 500,
+        sort: list[str] = None,
+        search: list[str] = None,
+    ) -> list[Book]:
         """Get list of books from calibre database.
 
         Args:
             limit (int): Limit on total number of results
+            sort (list[str]): List of strings to sort results by. Defaults to id
+                only
+            search (str): Search term
 
         Returns:
             list[Book]: List of books
         """
         if limit <= 0:
-            raise ValueError(f"limit {limit} not allowed")
+            raise ValueError(f"{limit=} cannot be <= 0")
 
         cmd = (
             f"{self.cdb_with_lib} list "
             f"--for-machine --fields=all "
             f"--limit={str(limit)}"
         )
+
+        cmd = self._handle_sort(cmd, sort)
+        cmd = self._handle_search(cmd, search)
+
         out, _ = self._run(cmd)
 
         books = json.loads(out)
-        res = []
-        if len(books):
-            for b in books:
-                res.append(Book(**b))
-        return res
+        if not len(books):
+            return []
+
+        return [Book(**b) for b in books]
+
+    def _handle_sort(self, cmd: str, sort: list[str]) -> str:
+        """Handle sort.
+
+        Unlike calibredb, this will default to ascending sort, unless a `-` is
+        prepended to ANY sort keys. Sort keys that are not supported are dropped
+        with a warning.
+
+        Args:
+            cmd (str): Command string to run
+            sort (list[str]): List of sort keys
+
+        Returns:
+            str: Command string with sort flags
+        """
+        if sort is None or not len(sort):
+            # default to ascending
+            cmd += " --ascending"
+            return cmd
+
+        # filter for unsupported sort keys
+        safe_sort = [x for x in sort if x.removeprefix("-") in self.SORT_BY_KEYS]
+        unsafe_sort = [x for x in sort if x not in safe_sort]
+        if len(unsafe_sort):
+            self.logger.warning(
+                f"The following sort keys are not supported and will be ignored: "
+                f"\"{', '.join(unsafe_sort)}\"."
+            )
+
+        descending = any(map(lambda x: x.startswith("-"), safe_sort))
+        if not descending:
+            cmd += " --ascending"
+
+        if len(safe_sort):
+            safe_sort = [x.removeprefix("-") for x in safe_sort]
+            safe_sort = ",".join(safe_sort)
+            cmd += f" --sort-by={safe_sort}"
+
+        return cmd
+
+    def _handle_search(self, cmd: str, search: list[str]) -> str:
+        if search is None or not len(search):
+            return cmd
+
+        cmd += f' --search "{" ".join(search)}"'
+        return cmd
 
     def add_one(
         self, book_path: str, book: Book = None, automerge: str = "ignore"
@@ -413,7 +492,7 @@ def _handle_add_flags(self, cmd: str, book: Book = None) -> str:
         if book is None:
             return cmd
 
-        for flag in self.ADD_FLAGS.keys():
+        for flag in self.ADD_FLAGS:
             value = getattr(book, flag)
             if value:
                 flag_name = self.ADD_FLAGS[flag]
@@ -448,7 +527,7 @@ def remove(self, ids: list[int], permanent: bool = False) -> str:
             str: Stdout of command, which is usually empty.
         """
         if not all(i >= 0 for i in ids):
-            raise ValueError(f"ids {ids} not allowed")
+            raise ValueError(f"{ids=} not allowed")
 
         cmd = f'{self.cdb_with_lib} remove {",".join(map(str, ids))}'
         if permanent:
@@ -603,4 +682,4 @@ def quote(s: str) -> str:
 
 def validate_id(id: int) -> None:
     if id <= 0:
-        raise ValueError(f"Value {id} cannot be <= 0")
+        raise ValueError(f"Value {id=} cannot be <= 0")

calibre_rest/models.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass, field
+from urllib.parse import urlencode, urlsplit, urlunsplit
 
 from jsonschema import Draft202012Validator
 
@@ -88,58 +89,82 @@ def validate(cls, instance):
 
 
 class PaginatedResults:
-    """
+    """Paginate list of books with offset and limit.
+
     Fields:
-    books (list[Book]): List of books
-
-    {
-        "books": [...]
-        "metadata": {
-            "total_count": 200
-            "limit": 10
-            "self": "/books?limit=10&after_id=0"
-            "prev": ""
-            "next": "/books?limit=10&after_id=10"
-        }
-    }
+        books (list[Book]): Full list of books
+        start (int): Start index
+        limit (int): Number of books per page
+        sort (list[str]): List of sort keys
+        search (list[str]): List of search terms
     """
 
-    def __init__(self, books: list[Book], total_count: int, limit: int):
+    def __init__(
+        self,
+        books: list[Book],
+        start: int,
+        limit: int,
+        sort: list[str] = None,
+        search: list[str] = None,
+    ):
+        self.base_url = urlsplit("/books")
         self.books = books
-        self.total_count = total_count
+        self.start = start
         self.limit = limit
-        self.before_id = -1
-        self.after_id = 0
+        self.sort = sort
+        self.search = search
+
+        if len(books) < self.start:
+            raise Exception(
+                f"start {self.start} is larger than number of books ({len(books)})"
+            )
+        self.count = len(books)
+
+    def build_query(self, start: int):
+        params = {"start": start, "limit": self.limit}
+
+        if self.sort is not None:
+            params["sort"] = self.sort
+
+        if self.search is not None:
+            params["search"] = self.search
+
+        query = urlencode(params, doseq=True)
+        return urlunsplit(self.base_url._replace(query=query))
 
     def current_page(self):
-        pass
+        return self.build_query(self.start)
 
     def prev_page(self):
         if not self.has_prev_page():
             return ""
 
-        return f"/books?limit={self.limit}&before_id={self.after_id - self.limit}"
+        prev_start = max(1, self.start - self.limit)
+        return self.build_query(prev_start)
 
     def next_page(self):
         if not self.has_next_page():
             return ""
 
-        return f"/books?limit={self.limit}&after_id={self.limit + self.after_id}"
+        return self.build_query(self.start + self.limit)
 
     def has_prev_page(self):
-        return self.before_id < 0
+        return not (self.start == 1)
 
     def has_next_page(self):
-        return not (self.limit + self.after_id > self.total_count)
+        return not (self.start + self.limit > self.count)
 
     def todict(self):
         return {
-            "books": self.books,
+            "books": self.books[
+                (self.start - 1) : (self.start - 1 + self.limit)  # noqa
+            ],
             "metadata": {
-                "total_count": self.total_count,
+                "start": self.start,
                 "limit": self.limit,
-                "self": f"{self.current_page()}",
-                "prev": f"{self.prev_page()}",
-                "next": f"{self.next_page()}",
+                "count": self.count,
+                "self": self.current_page(),
+                "prev": self.prev_page(),
+                "next": self.next_page(),
             },
         }

calibre_rest/routes.py

@@ -14,7 +14,7 @@
     ExistingItemError,
     InvalidPayloadError,
 )
-from calibre_rest.models import Book
+from calibre_rest.models import Book, PaginatedResults
 
 calibredb = app.config["CALIBRE_WRAPPER"]
 
@@ -43,24 +43,38 @@ def get_books():
     """Get list of books.
 
     Query Parameters:
-    limit: Maximum number of results in a page
-    before_id: Results before id
-    after_id: Results after id
-    sort: Sort results by field
-    search: Search results
+        start: Start index
+        limit: Maximum number of results in a page
+        sort: Sort results by field
+        search: Search results with Calibre's search interface
     """
 
-    per_page = request.args.get("per_page")
+    start = request.args.get("start") or 1
+    limit = request.args.get("limit") or 500
+    sort = request.args.getlist("sort") or None
+    search = request.args.getlist("search") or None
 
-    if per_page:
-        books = calibredb.get_books(limit=int(per_page))
-    else:
-        books = calibredb.get_books()
+    # Currently, calibredb fetches all books with the query and we perform the
+    # pagination on the results after. This means all results are fetched for
+    # every page query with no caching. This seems extremely wasteful, and the
+    # ideal way should be to perform pagination during the query, but calibredb
+    # does not support any form of offset or cursor based pagination of its
+    # results.
+
+    # Naively, we set a hard limit on the maximum number of results calibredb
+    # can fetch in one query, but there would no way to access the results that
+    # come after the limit with the same query.
 
+    books = calibredb.get_books(int(limit), sort, search)
     if not len(books):
         return response(204, jsonify(books=[]))
 
-    return response(200, jsonify(books=books))
+    try:
+        res = PaginatedResults(books, int(start), int(limit), sort, search)
+    except Exception as exc:
+        abort(400, exc)
+
+    return response(200, jsonify(res.todict()))
 
 
 # TODO add multiple and with directory

tests/calibre_test.py

@@ -29,6 +29,55 @@ def test_run_success(calibre):
     assert err == ""
 
 
+@pytest.mark.parametrize(
+    "keys, expected",
+    (
+        pytest.param(["id"], " --ascending --sort-by=id", id="ascending"),
+        pytest.param(["-id"], " --sort-by=id", id="descending"),
+        pytest.param(
+            ["title", "authors", "uuid"],
+            " --ascending --sort-by=title,authors,uuid",
+            id="multiple",
+        ),
+        pytest.param(
+            ["title", "authors", "-uuid"],
+            " --sort-by=title,authors,uuid",
+            id="multiple descending",
+        ),
+        pytest.param([], " --ascending", id="empty"),
+        pytest.param(["not_exist"], " --ascending", id="invalid"),
+        pytest.param(
+            ["title", "not_exist"],
+            " --ascending --sort-by=title",
+            id="mixed invalid",
+        ),
+    ),
+)
+def test_handle_sort(keys, expected):
+    cmd = "calibredb list"
+    got = dud_wrapper._handle_sort(cmd, keys)
+    assert got == "calibredb list" + expected
+
+
+@pytest.mark.parametrize(
+    "search, expected",
+    (
+        pytest.param(["title:foo"], ' --search "title:foo"', id="single"),
+        pytest.param(
+            ["title:foo", "id:5", "series:bar"],
+            ' --search "title:foo id:5 series:bar"',
+            id="multiple",
+        ),
+        pytest.param(["title:^f*"], ' --search "title:^f*"', id="regex"),
+        pytest.param([], "", id="empty"),
+    ),
+)
+def test_handle_search(search, expected):
+    cmd = "calibredb list"
+    got = dud_wrapper._handle_search(cmd, search)
+    assert got == "calibredb list" + expected
+
+
 @pytest.mark.parametrize(
     "book, expected",
     (