120 lines
4 KiB
Python
120 lines
4 KiB
Python
"""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({'detail': 'Invalid page.'}, status_code=HTTPStatus.NOT_FOUND)
|
|
|
|
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],
|
|
)
|