migrated to modern-rest

stratoflights_api/pagination.py

@@ -0,0 +1,120 @@
+"""Limit/offset pagination for the prediction list endpoints.
+
+DMR has no built-in limit/offset paginator, so (as the docs suggest) we keep our
+own envelope model -- matching the former ``CustomLimitOffsetPagination`` output
+exactly -- plus a helper that slices the queryset. The query params and envelope
+keys are preserved verbatim: ``limit`` / ``skip`` in, and
+``{total, limit, skip, predictions}`` out.
+"""
+from http import HTTPStatus
+from typing import Optional
+from urllib.parse import urlsplit, urlunsplit
+
+from django.core.paginator import InvalidPage, Paginator
+from django.http import QueryDict
+from pydantic import BaseModel
+
+from dmr.response import APIError
+
+from .dtos import PredictionOut, TelemetryOut
+
+DEFAULT_LIMIT = 10  # was CustomLimitOffsetPagination.default_limit
+MAX_LIMIT = 100     # was CustomLimitOffsetPagination.max_limit
+PAGE_SIZE = 100     # global REST_FRAMEWORK PAGE_SIZE (PageNumberPagination)
+
+
+class PaginationQuery(BaseModel):
+    # Param names preserved: limit_query_param='limit', offset_query_param='skip'.
+    limit: int = DEFAULT_LIMIT
+    skip: int = 0
+
+
+class PredictionListUserQuery(BaseModel):
+    """Query params for PredictionViewSet.list_user: filters + limit/offset."""
+
+    satellite_id: Optional[str] = None
+    created_from: Optional[str] = None
+    created_till: Optional[str] = None
+    limit: int = DEFAULT_LIMIT
+    skip: int = 0
+
+
+class PredictionPage(BaseModel):
+    total: int
+    limit: int
+    skip: int
+    predictions: list[PredictionOut]
+
+
+class TelemetryPage(BaseModel):
+    count: int
+    next: Optional[str] = None
+    previous: Optional[str] = None
+    results: list[TelemetryOut]
+
+
+def _replace_query_param(url: str, key: str, value) -> str:
+    scheme, netloc, path, query, fragment = urlsplit(url)
+    query_dict = QueryDict(query, mutable=True)
+    query_dict[key] = value
+    return urlunsplit((scheme, netloc, path, query_dict.urlencode(), fragment))
+
+
+def _remove_query_param(url: str, key: str) -> str:
+    scheme, netloc, path, query, fragment = urlsplit(url)
+    query_dict = QueryDict(query, mutable=True)
+    query_dict.pop(key, None)
+    return urlunsplit((scheme, netloc, path, query_dict.urlencode(), fragment))
+
+
+def page_number_paginate(request, queryset, page_size: int = PAGE_SIZE):
+    """Reproduce DRF PageNumberPagination.
+
+    Returns ``(count, next_link, previous_link, object_list)``. Raises a 404
+    "Invalid page." like DRF on an out-of-range / non-integer page. Honours
+    ``page=last`` and builds absolute next/previous links off the current URL.
+    """
+    paginator = Paginator(queryset, page_size)
+    page_number = request.GET.get('page', 1)
+    if page_number in ('last',):
+        page_number = paginator.num_pages
+    try:
+        page = paginator.page(page_number)
+    except InvalidPage:
+        raise APIError(status_code=HTTPStatus.NOT_FOUND, body={'detail': 'Invalid page.'})
+
+    url = request.build_absolute_uri()
+
+    next_link = None
+    if page.has_next():
+        next_link = _replace_query_param(url, 'page', page.next_page_number())
+
+    previous_link = None
+    if page.has_previous():
+        previous_number = page.previous_page_number()
+        if previous_number == 1:
+            previous_link = _remove_query_param(url, 'page')
+        else:
+            previous_link = _replace_query_param(url, 'page', previous_number)
+
+    return paginator.count, next_link, previous_link, list(page.object_list)
+
+
+def paginate_predictions(queryset, query: PaginationQuery) -> PredictionPage:
+    """Slice ``queryset`` and build the prediction page envelope.
+
+    Mirrors the bounds DRF's LimitOffsetPagination enforced: limit falls back to
+    the default when non-positive and is capped at MAX_LIMIT; skip floors at 0.
+    """
+    limit = query.limit if query.limit > 0 else DEFAULT_LIMIT
+    limit = min(limit, MAX_LIMIT)
+    skip = query.skip if query.skip >= 0 else 0
+
+    total = queryset.count()
+    page = list(queryset[skip:skip + limit])
+    return PredictionPage(
+        total=total,
+        limit=limit,
+        skip=skip,
+        predictions=[PredictionOut.model_validate(obj) for obj in page],
+    )