Add methods to get URLs from a PaginatedList

Add separate methods to return the first, next, and previous URLs
for a paginated list to make the HATEOS approach a bit easier, and
implement `link_header` in terms of those methods.

docs/user-guide/database/pagination.rst

@@ -190,6 +190,9 @@ Either way, the results will be a `~safir.database.PaginatedList` wrapping a lis
 Returning paginated results
 ===========================
 
+Using the HTTP headers
+----------------------
+
 HTTP provides the ``Link`` header (:rfc:`8288`) to declare relationships between multiple web responses.
 Using a ``Link`` header with relation types ``first``, ``next``, and ``prev`` is a standard way of providing the client with pagination information.
 
@@ -232,3 +235,12 @@ A real route handler would have more query parameters and more documentation.
 Note that this example also sets a non-standard ``X-Total-Count`` header containing the total count of entries returned by the underlying query without pagination.
 `~safir.database.PaginatedQueryRunner` obtains this information by default, since the count query is often fast for databases to perform.
 There is no standard way to return this information to the client, but ``X-Total-Count`` is a widely-used informal standard.
+
+Including links in the response
+-------------------------------
+
+Alternately, some web services may instead wish to return the paginated results inside a JSON data structure that includes the pagination information.
+This follows the `HATEOS <https://en.wikipedia.org/wiki/HATEOAS>`__ design principle of embedding links inside the returned data.
+
+In this case, the application should call the `~safir.database.PaginatedList.first_url`, `~safir.database.PaginatedList.next_url`, and `~safir.database.PaginatedList.prev_url` methods with the current URL (generally ``request.url``) as an argument to retrieve the links to the first, next, and previous blocks of results.
+Those links can then be embedded in the response model wherever is appropriate for the API of that application.

safir/src/safir/database/_pagination.py

@@ -278,24 +278,83 @@ class PaginatedList(Generic[E, C]):
     prev_cursor: C | None = None
     """Cursor for the previous batch of entries."""
 
-    def link_header(self, base_url: URL) -> str:
+    def first_url(self, current_url: URL) -> str:
+        """Construct a URL to the first group of results for this query.
+
+        Parameters
+        ----------
+        current_url
+            The starting URL of the current group of entries.
+
+        Returns
+        -------
+        str
+            URL to the first group of entries for this query.
+        """
+        return str(current_url.remove_query_params("cursor"))
+
+    def next_url(self, current_url: URL) -> str | None:
+        """Construct a URL to the next group of results for this query.
+
+        Parameters
+        ----------
+        current_url
+            The starting URL of the current group of entries.
+
+        Returns
+        -------
+        str or None
+            URL to the next group of entries for this query or `None` if there
+            are no further entries.
+        """
+        if not self.next_cursor:
+            return None
+        first_url = current_url.remove_query_params("cursor")
+        params = parse_qs(first_url.query)
+        params["cursor"] = [str(self.next_cursor)]
+        return str(first_url.replace(query=urlencode(params, doseq=True)))
+
+    def prev_url(self, current_url: URL) -> str | None:
+        """Construct a URL to the previous group of results for this query.
+
+        Parameters
+        ----------
+        current_url
+            The starting URL of the current group of entries.
+
+        Returns
+        -------
+        str or None
+            URL to the previous group of entries for this query or `None` if
+            there are no further entries.
+        """
+        if not self.prev_cursor:
+            return None
+        first_url = current_url.remove_query_params("cursor")
+        params = parse_qs(first_url.query)
+        params["cursor"] = [str(self.prev_cursor)]
+        return str(first_url.replace(query=urlencode(params, doseq=True)))
+
+    def link_header(self, current_url: URL) -> str:
         """Construct an RFC 8288 ``Link`` header for a paginated result.
 
         Parameters
         ----------
-        base_url
+        current_url
             The starting URL of the current group of entries.
+
+        Returns
+        -------
+        str
+            Contents of an RFC 8288 ``Link`` header.
         """
-        first_url = base_url.remove_query_params("cursor")
+        first_url = self.first_url(current_url)
+        next_url = self.next_url(current_url)
+        prev_url = self.prev_url(current_url)
         header = f'<{first_url!s}>; rel="first"'
-        params = parse_qs(first_url.query)
-        if self.next_cursor:
-            params["cursor"] = [str(self.next_cursor)]
-            next_url = first_url.replace(query=urlencode(params, doseq=True))
+        if next_url:
             header += f', <{next_url!s}>; rel="next"'
-        if self.prev_cursor:
-            params["cursor"] = [str(self.prev_cursor)]
-            prev_url = first_url.replace(query=urlencode(params, doseq=True))
+        if prev_url:
             header += f', <{prev_url!s}>; rel="prev"'
         return header
 

safir/tests/database_test.py

@@ -418,10 +418,13 @@ async def test_pagination(database_url: str, database_password: str) -> None:
         assert result.count == 7
         assert not result.prev_cursor
         base_url = URL("https://example.com/query")
+        next_url = f"{base_url!s}?cursor={result.next_cursor}"
         assert result.link_header(base_url) == (
-            f'<{base_url!s}>; rel="first", '
-            f'<{base_url!s}?cursor={result.next_cursor}>; rel="next"'
+            f'<{base_url!s}>; rel="first", ' f'<{next_url}>; rel="next"'
         )
+        assert result.first_url(base_url) == str(base_url)
+        assert result.next_url(base_url) == next_url
+        assert result.prev_url(base_url) is None
         assert str(result.next_cursor) == "1600000000.5_1"
 
         result = await builder.query_object(
@@ -434,13 +437,18 @@ async def test_pagination(database_url: str, database_password: str) -> None:
         assert result.count == 7
         assert str(result.next_cursor) == "1510000000_2"
         assert str(result.prev_cursor) == "p1600000000.5_1"
-        base_url = URL("https://example.com/query?foo=bar&cursor=xxxx")
-        stripped_url = "https://example.com/query?foo=bar"
+        base_url = URL("https://example.com/query?foo=bar&foo=baz&cursor=xxxx")
+        stripped_url = "https://example.com/query?foo=bar&foo=baz"
+        next_url = f"{stripped_url}&cursor={result.next_cursor}"
+        prev_url = f"{stripped_url}&cursor={result.prev_cursor}"
         assert result.link_header(base_url) == (
             f'<{stripped_url}>; rel="first", '
-            f'<{stripped_url}&cursor={result.next_cursor}>; rel="next", '
-            f'<{stripped_url}&cursor={result.prev_cursor}>; rel="prev"'
+            f'<{next_url}>; rel="next", '
+            f'<{prev_url}>; rel="prev"'
         )
+        assert result.first_url(base_url) == stripped_url
+        assert result.next_url(base_url) == next_url
+        assert result.prev_url(base_url) == prev_url
         next_cursor = result.next_cursor
 
         result = await builder.query_object(
@@ -461,6 +469,7 @@ async def test_pagination(database_url: str, database_password: str) -> None:
         assert result.count == 7
         assert not result.next_cursor
         base_url = URL("https://example.com/query")
+        assert result.next_url(base_url) is None
         assert result.link_header(base_url) == (
             f'<{base_url!s}>; rel="first", '
             f'<{base_url!s}?cursor={result.prev_cursor}>; rel="prev"'