Keep every filter in pagination links; fix previous past the end

app/routers/search.py

@@ -1,8 +1,7 @@
 import csv
 from contextlib import aclosing
 from tempfile import SpooledTemporaryFile
-from typing import Any, AsyncIterator, Dict, Iterator, List, Literal, Optional
-from urllib.parse import urlencode
+from typing import Any, AsyncIterator, Dict, Iterator, List, Literal, Optional, Tuple
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from fastapi.exceptions import RequestValidationError
@@ -73,6 +72,24 @@ def read_chunks(file: SpooledTemporaryFile, size: int = 64 * 1024) -> Iterator[s
         yield chunk
 
 
+def page_links(request: Request, offset: int, limit: int, total: int) -> Tuple[Optional[str], Optional[str]]:
+    """Build next and previous links: the request's own URL with offset and limit replaced.
+
+    Reusing the request URL keeps every parameter under the name the caller sent.
+    """
+    next_url = None
+    if offset + limit < total:
+        next_url = str(request.url.include_query_params(offset=offset + limit, limit=limit))
+
+    previous_url = None
+    if offset > 0:
+        # Past the end, go back to the last page instead of another empty one
+        prev_offset = max(0, min(offset - limit, total - limit))
+        previous_url = str(request.url.include_query_params(offset=prev_offset, limit=limit))
+
+    return next_url, previous_url
+
+
 def parse_query_params(
     # String filters
     accession: Optional[List[str]] = Query(
@@ -335,46 +352,9 @@ async def get_data(
 
             # Add pagination links
             if request:
-                base_url = str(request.url).split("?")[0]
-
-                # Prepare query parameters for pagination links
-                query_params = {
-                    k: v
-                    for k, v in params.model_dump().items()
-                    if k not in ["auto_paginated", "offset", "limit"]
-                    and v is not None
-                }
-
-                # Set format explicitly if it was provided
-                if params.format != ResponseFormat.JSON:
-                    query_params["format"] = params.format
-
-                current_offset = params.offset or 0
-                current_limit = params.limit
-
-                # Next page link if there are more records
-                if current_offset + current_limit < total_count:
-                    next_offset = current_offset + current_limit
-                    next_params = {
-                        **query_params,
-                        "offset": next_offset,
-                        "limit": current_limit,
-                    }
-                    response.next = (
-                        f"{base_url}?{urlencode(next_params, doseq=True)}"
-                    )
-
-                # Previous page link if not on first page
-                if current_offset > 0:
-                    prev_offset = max(0, current_offset - current_limit)
-                    prev_params = {
-                        **query_params,
-                        "offset": prev_offset,
-                        "limit": current_limit,
-                    }
-                    response.previous = (
-                        f"{base_url}?{urlencode(prev_params, doseq=True)}"
-                    )
+                response.next, response.previous = page_links(
+                    request, params.offset or 0, params.limit, total_count
+                )
 
             return response
 
@@ -567,47 +547,7 @@ async def get_typeahead(
         previous_url = None
 
         if request:
-            base_url = str(request.url).split("?")[0]
-
-            # Build base query params (excluding pagination)
-            base_params = {
-                "field_name": params.field_name,
-                "search": params.search,
-                "limit": limit,
-            }
-
-            # Add context params if present
-            if params.accession:
-                base_params["accession"] = params.accession
-            if params.organism:
-                base_params["organism"] = params.organism
-            if params.protein_name:
-                base_params["protein"] = params.protein_name
-            if params.gene_name:
-                base_params["gene_name"] = params.gene_name
-            if params.clean_ec_number:
-                base_params["ec_number"] = params.clean_ec_number
-            if params.uniprot_id:
-                base_params["uniprot"] = params.uniprot_id
-            if params.curation_status:
-                base_params["curation_status"] = params.curation_status
-            if params.clean_ec_confidence_min is not None:
-                base_params["clean_ec_confidence_min"] = params.clean_ec_confidence_min
-            if params.clean_ec_confidence_max is not None:
-                base_params["clean_ec_confidence_max"] = params.clean_ec_confidence_max
-            if params.sequence_length:
-                base_params["sequence_length"] = params.sequence_length
-
-            # Next page
-            if offset + limit < total:
-                next_params = {**base_params, "offset": offset + limit}
-                next_url = f"{base_url}?{urlencode(next_params, doseq=True)}"
-
-            # Previous page
-            if offset > 0:
-                prev_offset = max(0, offset - limit)
-                prev_params = {**base_params, "offset": prev_offset}
-                previous_url = f"{base_url}?{urlencode(prev_params, doseq=True)}"
+            next_url, previous_url = page_links(request, offset, limit, total)
 
         return CLEANTypeaheadResponse(
             field_name=params.field_name,

tests/test_search_pagination_links.py

@@ -0,0 +1,98 @@
+import re
+from urllib.parse import parse_qs, urlsplit
+
+# One of every /search filter, using the query parameter names clients send
+FILTERS = (
+    "accession=P51451&organism=Homo%20sapiens&organism=Mus%20musculus&protein=Kinase"
+    "&gene_name=BLK&ec_number=2.7.10.2&uniprot=BLK_HUMAN&curation_status=reviewed"
+    "&clean_ec_confidence_min=0.5&clean_ec_confidence_max=0.9&sequence_length=100"
+    "&ordering=-accession"
+)
+
+# One of every /typeahead search context filter
+TYPEAHEAD_FILTERS = (
+    "field_name=organism&search=homo&accession=P51451&organism=Homo%20sapiens&protein=Kinase"
+    "&gene_name=BLK&ec_number=2.7.10.2&uniprot=BLK_HUMAN&curation_status=reviewed"
+    "&clean_ec_confidence_min=0.5&clean_ec_confidence_max=0.9&sequence_length=100"
+)
+
+
+def queries(db):
+    """The SQL and arguments of the last COUNT and data queries, ignoring LIMIT and OFFSET."""
+    def normalize(call):
+        sql = re.sub(r"\b(LIMIT|OFFSET) \d+", r"\1 _", call.args[0])
+        return " ".join(sql.split()), call.args[1:]
+
+    return normalize(db.fetchval.call_args), normalize(db.fetch.call_args)
+
+
+def follow(client, db, url):
+    """Request url and return the queries it ran. Fails if it never reached the DB."""
+    db.fetchval.reset_mock()
+    db.fetch.reset_mock()
+    parts = urlsplit(url)
+    response = client.get(f"{parts.path}?{parts.query}")
+    assert response.status_code == 200, response.text
+    return queries(db)
+
+
+def link_params(url):
+    return parse_qs(urlsplit(url).query)
+
+
+def test_next_and_previous_keep_every_filter(client, db):
+    db.fetchval.return_value = 100
+    response = client.get(f"/api/v1/search?{FILTERS}&limit=10&offset=20")
+    original_queries = queries(db)
+    original = parse_qs(FILTERS)
+
+    links = {name: response.json()[name] for name in ("next", "previous")}
+    for name, offset in (("next", "30"), ("previous", "10")):
+        assert link_params(links[name]) == {**original, "limit": ["10"], "offset": [offset]}, name
+        assert follow(client, db, links[name]) == original_queries, name
+
+
+def test_next_uses_the_default_page_size_when_no_limit_given(client, db):
+    db.fetchval.return_value = 10_000
+    response = client.get("/api/v1/search?ec_number=1.1.1.1")
+
+    assert link_params(response.json()["next"]) == {
+        "ec_number": ["1.1.1.1"], "limit": ["5000"], "offset": ["5000"],
+    }
+
+
+def test_previous_past_the_end_goes_to_the_last_page(client, db):
+    db.fetchval.return_value = 36_682
+    body = client.get("/api/v1/search?ec_number=1.1.1.1&limit=10&offset=100000").json()
+
+    assert body["next"] is None
+    assert link_params(body["previous"])["offset"] == ["36672"]
+
+
+def test_previous_past_the_end_of_no_results_goes_to_the_start(client, db):
+    db.fetchval.return_value = 0
+    body = client.get("/api/v1/search?ec_number=1.1.1.1&limit=10&offset=50").json()
+
+    assert link_params(body["previous"])["offset"] == ["0"]
+
+
+def test_typeahead_next_and_previous_keep_every_filter(client, db):
+    db.fetchval.return_value = 100
+    response = client.get(f"/api/v1/typeahead?{TYPEAHEAD_FILTERS}&limit=10&offset=20")
+    original_queries = queries(db)
+    original = parse_qs(TYPEAHEAD_FILTERS)
+
+    links = {name: response.json()[name] for name in ("next", "previous")}
+    for name, offset in (("next", "30"), ("previous", "10")):
+        assert link_params(links[name]) == {**original, "limit": ["10"], "offset": [offset]}, name
+        assert follow(client, db, links[name]) == original_queries, name
+
+
+def test_typeahead_previous_past_the_end_goes_to_the_last_page(client, db):
+    db.fetchval.return_value = 50
+    body = client.get("/api/v1/typeahead?field_name=organism&search=homo&offset=1000").json()
+
+    assert body["next"] is None
+    assert link_params(body["previous"]) == {
+        "field_name": ["organism"], "search": ["homo"], "limit": ["20"], "offset": ["30"],
+    }