107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""Pagination metadata from Gitea response headers.
|
|
|
|
Gitea paginates with ``page``/``limit`` query parameters and reports the
|
|
neighbours in a ``Link`` header plus the overall count in ``X-Total-Count``.
|
|
Only the page numbers in the links are used; the link URLs themselves are never
|
|
followed or returned (they are untrusted).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
_LINK_ENTRY = re.compile(r"<([^>]*)>([^<]*)")
|
|
_REL = re.compile(r'rel\s*=\s*"?([^";,]+)"?', re.IGNORECASE)
|
|
|
|
|
|
def _int(value: Any, *, minimum: int = 1) -> int | None:
|
|
"""Return ``value`` as an int of at least ``minimum``, or None."""
|
|
try:
|
|
number = int(str(value).strip())
|
|
except ValueError:
|
|
return None
|
|
return number if number >= minimum else None
|
|
|
|
|
|
def parse_link_header(header: str | None) -> dict[str, dict[str, int | None]]:
|
|
"""Parse a ``Link`` header into ``{rel: {"page": n, "limit": n}}``.
|
|
|
|
Entries without a ``rel`` are ignored. Values that are not positive
|
|
integers become None.
|
|
"""
|
|
links: dict[str, dict[str, int | None]] = {}
|
|
for url, params in _LINK_ENTRY.findall(header or ""):
|
|
rel_match = _REL.search(params)
|
|
if rel_match is None:
|
|
continue
|
|
query = parse_qs(urlsplit(url).query)
|
|
numbers = {
|
|
key: _int(query[key][0]) if query.get(key) else None for key in ("page", "limit")
|
|
}
|
|
for rel in rel_match.group(1).split():
|
|
links[rel.lower()] = numbers
|
|
return links
|
|
|
|
|
|
def build_pagination(
|
|
headers: Mapping[str, str],
|
|
*,
|
|
requested_page: Any,
|
|
requested_limit: Any,
|
|
paginated: bool,
|
|
) -> dict[str, int | None] | None:
|
|
"""Build the envelope ``pagination`` object, or None when there is nothing to report.
|
|
|
|
Args:
|
|
headers: Response headers (case-insensitive mapping).
|
|
requested_page: ``page`` argument of the request, if any.
|
|
requested_limit: ``limit`` argument of the request, if any.
|
|
paginated: True when the operation supports pagination.
|
|
|
|
Returns:
|
|
``{"page", "limit", "total_count", "next_page", "last_page"}`` with None
|
|
for unknown values, or None when the operation is not paginated and the
|
|
response carries neither ``Link`` nor ``X-Total-Count``.
|
|
"""
|
|
link_header = headers.get("link")
|
|
total_header = headers.get("x-total-count")
|
|
if not paginated and link_header is None and total_header is None:
|
|
return None
|
|
|
|
links = parse_link_header(link_header)
|
|
total_count = _int(total_header, minimum=0)
|
|
|
|
next_page = links.get("next", {}).get("page")
|
|
prev_page = links.get("prev", {}).get("page")
|
|
last_page = links.get("last", {}).get("page")
|
|
|
|
page = _int(requested_page)
|
|
if page is None:
|
|
if prev_page is not None:
|
|
page = prev_page + 1
|
|
elif next_page is not None:
|
|
page = next_page - 1 or 1
|
|
else:
|
|
page = 1
|
|
|
|
limit = _int(requested_limit)
|
|
if limit is None:
|
|
for rel in ("next", "last", "prev", "first"):
|
|
limit = links.get(rel, {}).get("limit")
|
|
if limit is not None:
|
|
break
|
|
|
|
if next_page is None and last_page is None:
|
|
last_page = page
|
|
|
|
return {
|
|
"page": page,
|
|
"limit": limit,
|
|
"total_count": total_count,
|
|
"next_page": next_page,
|
|
"last_page": last_page,
|
|
}
|