Pagination contract is not standardized across endpoints

This commit is contained in:
2026-04-28 21:40:22 +02:00
parent ad21590f60
commit a2129bb9bd
13 changed files with 387 additions and 56 deletions

View File

@@ -10,7 +10,7 @@ from enum import StrEnum
from pydantic import AnyHttpUrl, Field
from app.models.response import BanGuiBaseModel
from app.models.response import BanGuiBaseModel, PaginatedListResponse
# ---------------------------------------------------------------------------
@@ -69,14 +69,14 @@ class ImportLogEntry(BanGuiBaseModel):
ips_skipped: int
errors: str | None
class ImportLogListResponse(BanGuiBaseModel):
"""Response for ``GET /api/blocklists/log``."""
class ImportLogListResponse(PaginatedListResponse[ImportLogEntry]):
"""Response for ``GET /api/blocklists/log``.
items: list[ImportLogEntry] = Field(default_factory=list)
total: int = Field(..., ge=0)
page: int = Field(default=1, ge=1)
page_size: int = Field(default=50, ge=1)
total_pages: int = Field(default=1, ge=1)
Paginated list of all blocklist import runs with timestamps, source info,
and per-source import/skip counts.
"""
pass
# ---------------------------------------------------------------------------
# Schedule

View File

@@ -7,7 +7,7 @@ from __future__ import annotations
from pydantic import Field
from app.models.response import BanGuiBaseModel
from app.models.response import BanGuiBaseModel, PaginatedListResponse
from app.models.ban import TimeRange
@@ -56,13 +56,15 @@ class HistoryBanItem(BanGuiBaseModel):
description="Organisation name associated with the IP.",
)
class HistoryListResponse(BanGuiBaseModel):
"""Paginated history ban-list response."""
class HistoryListResponse(PaginatedListResponse[HistoryBanItem]):
"""Paginated history ban-list response.
items: list[HistoryBanItem] = Field(default_factory=list)
total: int = Field(..., ge=0, description="Total matching records.")
page: int = Field(..., ge=1)
page_size: int = Field(..., ge=1)
Request: ``GET /api/history`` with optional time-range, jail, IP, and
origin filters plus pagination parameters.
Response: Paginated collection of historical ban records with geolocation.
"""
pass
# ---------------------------------------------------------------------------
# Per-IP timeline

View File

@@ -46,6 +46,7 @@ from app.models.blocklist import (
)
from app.services import ban_service, blocklist_service
from app.tasks.blocklist_import import run_import_with_resources
from app.utils.constants import DEFAULT_PAGE_SIZE
router: APIRouter = APIRouter(prefix="/api/blocklists", tags=["Blocklists"])
@@ -221,8 +222,10 @@ async def get_import_log(
blocklist_ctx: BlocklistServiceContextDep,
_auth: AuthDep,
source_id: int | None = Query(default=None, description="Filter by source id"),
page: int = Query(default=1, ge=1),
page_size: int = Query(default=50, ge=1, le=200),
page: int = Query(default=1, ge=1, description="1-based page number."),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE, ge=1, le=500, description="Items per page (max 500)."
),
) -> ImportLogListResponse:
"""Return a paginated log of all import runs.

View File

@@ -540,14 +540,12 @@ async def list_import_logs(
items, total = await import_log_repo.list_logs(
db, source_id=source_id, page=page, page_size=page_size
)
total_pages = import_log_repo.compute_total_pages(total, page_size)
return ImportLogListResponse(
items=[ImportLogEntry.model_validate(i) for i in items],
total=total,
page=page,
page_size=page_size,
total_pages=total_pages,
)

View File

@@ -0,0 +1,102 @@
"""Pagination utilities and standardized query parameter handling.
This module provides reusable utilities for implementing consistent pagination
across all endpoints. All paginated endpoints should use these utilities to
ensure a uniform API contract.
Standard Pagination Contract:
Query parameters: page (1-based), page_size (1-500)
Response: PaginatedListResponse[T] with items, total, page, page_size
Usage in routers:
```python
from app.utils.pagination import PAGINATION_DEFAULTS
@router.get("/items")
async def get_items(
page: int = Query(default=PAGINATION_DEFAULTS["page"], ge=1),
page_size: int = Query(
default=PAGINATION_DEFAULTS["page_size"],
ge=1,
le=PAGINATION_DEFAULTS["max_page_size"],
),
):
...
```
"""
from typing import Final
__all__ = ["PAGINATION_DEFAULTS", "get_offset", "compute_total_pages"]
# Standardized pagination defaults
PAGINATION_DEFAULTS: Final[dict[str, int]] = {
"page": 1,
"page_size": 100,
"max_page_size": 500,
}
def get_offset(page: int, page_size: int) -> int:
"""Calculate the database offset for a given page and page size.
Args:
page: 1-based page number.
page_size: Items per page.
Returns:
0-based database offset (number of items to skip).
Raises:
ValueError: If page or page_size is invalid (< 1).
Example:
```python
# Page 1, size 10 → offset 0
assert get_offset(1, 10) == 0
# Page 2, size 10 → offset 10
assert get_offset(2, 10) == 10
# Page 3, size 50 → offset 100
assert get_offset(3, 50) == 100
```
"""
if page < 1:
raise ValueError(f"page must be >= 1, got {page}")
if page_size < 1:
raise ValueError(f"page_size must be >= 1, got {page_size}")
return (page - 1) * page_size
def compute_total_pages(total: int, page_size: int) -> int:
"""Calculate the total number of pages needed.
Args:
total: Total number of items across all pages.
page_size: Items per page.
Returns:
The number of pages required to hold all items. Always at least 1,
even when total is 0 (an empty page is still a page).
Raises:
ValueError: If page_size is invalid (< 1).
Example:
```python
assert compute_total_pages(0, 10) == 1
assert compute_total_pages(10, 10) == 1
assert compute_total_pages(11, 10) == 2
assert compute_total_pages(25, 5) == 5
```
"""
if page_size < 1:
raise ValueError(f"page_size must be >= 1, got {page_size}")
if total == 0:
return 1
# Ceiling division: (total + page_size - 1) // page_size
return (total + page_size - 1) // page_size

View File

@@ -85,7 +85,7 @@ def _make_import_result() -> ImportRunResult:
def _make_log_response() -> ImportLogListResponse:
return ImportLogListResponse(
items=[], total=0, page=1, page_size=50, total_pages=1
items=[], total=0, page=1, page_size=50
)
@@ -457,10 +457,10 @@ class TestImportLog:
assert resp.status_code == 200
async def test_log_response_shape(self, bl_client: AsyncClient) -> None:
"""Log response has items, total, page, page_size, total_pages."""
"""Log response has items, total, page, page_size."""
resp = await bl_client.get("/api/blocklists/log")
body = resp.json()
for key in ("items", "total", "page", "page_size", "total_pages"):
for key in ("items", "total", "page", "page_size"):
assert key in body
async def test_log_empty_when_no_runs(self, bl_client: AsyncClient) -> None:

View File

@@ -529,7 +529,6 @@ class TestImportLogPagination:
assert resp.total == 0
assert resp.page == 1
assert resp.page_size == 10
assert resp.total_pages == 1
async def test_list_import_logs_paginates(self, db: aiosqlite.Connection) -> None:
"""list_import_logs computes total pages and returns the correct subset."""
@@ -549,7 +548,6 @@ class TestImportLogPagination:
db, source_id=None, page=2, page_size=2
)
assert resp.total == 3
assert resp.total_pages == 2
assert resp.page == 2
assert resp.page_size == 2
assert len(resp.items) == 1

View File

@@ -0,0 +1,104 @@
"""Tests for pagination utilities.
Validates the pagination helper functions used across all paginated endpoints.
"""
import pytest
from app.utils.pagination import get_offset, compute_total_pages, PAGINATION_DEFAULTS
class TestPaginationDefaults:
"""Test pagination default constants."""
def test_pagination_defaults_structure(self) -> None:
"""PAGINATION_DEFAULTS contains required keys."""
required_keys = {"page", "page_size", "max_page_size"}
assert required_keys.issubset(PAGINATION_DEFAULTS.keys())
def test_pagination_defaults_values(self) -> None:
"""PAGINATION_DEFAULTS have expected values."""
assert PAGINATION_DEFAULTS["page"] == 1
assert PAGINATION_DEFAULTS["page_size"] == 100
assert PAGINATION_DEFAULTS["max_page_size"] == 500
class TestGetOffset:
"""Test get_offset calculation."""
def test_first_page(self) -> None:
"""First page (page=1) produces offset=0."""
assert get_offset(1, 10) == 0
def test_second_page(self) -> None:
"""Second page (page=2) produces offset=page_size."""
assert get_offset(2, 10) == 10
def test_arbitrary_page(self) -> None:
"""Arbitrary pages produce correct offsets."""
assert get_offset(3, 50) == 100
assert get_offset(5, 25) == 100
assert get_offset(10, 100) == 900
def test_page_size_variations(self) -> None:
"""Offsets scale correctly with page_size."""
assert get_offset(2, 1) == 1
assert get_offset(2, 10) == 10
assert get_offset(2, 100) == 100
assert get_offset(2, 500) == 500
def test_invalid_page_raises(self) -> None:
"""Invalid page numbers raise ValueError."""
with pytest.raises(ValueError, match="page must be >= 1"):
get_offset(0, 10)
with pytest.raises(ValueError, match="page must be >= 1"):
get_offset(-1, 10)
def test_invalid_page_size_raises(self) -> None:
"""Invalid page sizes raise ValueError."""
with pytest.raises(ValueError, match="page_size must be >= 1"):
get_offset(1, 0)
with pytest.raises(ValueError, match="page_size must be >= 1"):
get_offset(1, -1)
class TestComputeTotalPages:
"""Test compute_total_pages calculation."""
def test_empty_collection(self) -> None:
"""Empty collection (total=0) produces at least 1 page."""
assert compute_total_pages(0, 10) == 1
def test_exact_page_fit(self) -> None:
"""Totals that fit exactly produce expected page count."""
assert compute_total_pages(10, 10) == 1
assert compute_total_pages(20, 10) == 2
assert compute_total_pages(50, 10) == 5
def test_partial_page(self) -> None:
"""Totals that overflow a page produce ceil(total / page_size)."""
assert compute_total_pages(11, 10) == 2
assert compute_total_pages(25, 5) == 5
assert compute_total_pages(101, 10) == 11
def test_single_item(self) -> None:
"""Single item produces 1 page regardless of page_size."""
assert compute_total_pages(1, 1) == 1
assert compute_total_pages(1, 10) == 1
assert compute_total_pages(1, 500) == 1
def test_large_page_size(self) -> None:
"""Large page sizes still compute correctly."""
assert compute_total_pages(1, 500) == 1
assert compute_total_pages(501, 500) == 2
assert compute_total_pages(1000, 500) == 2
def test_invalid_page_size_raises(self) -> None:
"""Invalid page sizes raise ValueError."""
with pytest.raises(ValueError, match="page_size must be >= 1"):
compute_total_pages(100, 0)
with pytest.raises(ValueError, match="page_size must be >= 1"):
compute_total_pages(100, -1)