woggioni
6932a3272c
CI / Build and push docker image (push) Successful in 3m12s
The platform now hosts multiple card games, with scopone scientifico as the first one. Rename the brand wherever it is not a game rule: - move the Python package to server/src/tavolo and update imports - rename the Postgres database/user, OIDC issuer path, client id and Redis key prefixes to tavolo (clean break: existing pgdata volumes and live games are not migrated) - rename the Cargo package to tavolo-web and set the page title to Tavolo - update docs and the Docker image path to woggioni/tavolo The scopa game term (clearing the table) in the engine, state and web UI is intentionally left untouched.
129 lines
4.0 KiB
Python
129 lines
4.0 KiB
Python
"""Cursor-based pagination for listing endpoints.
|
|
|
|
Uses keyset pagination (not OFFSET/LIMIT): each page ends with an opaque
|
|
cursor encoding the sort key tuple of the last item on that page; the next
|
|
request passes that cursor and the query continues from the point it left
|
|
off. This is stable under concurrent inserts and cheaper than OFFSET for
|
|
large result sets.
|
|
|
|
Cursor is a base64-encoded JSON object mapping the sort-field names to the
|
|
values of the last item on the previous page. It's opaque to callers and
|
|
must be treated as a black box.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from tortoise.queryset import QuerySet
|
|
|
|
from .http import extract_query_params
|
|
|
|
DEFAULT_LIMIT = 20
|
|
MAX_LIMIT = 100
|
|
MIN_LIMIT = 1
|
|
|
|
CURSOR_PARAM = "cursor"
|
|
LIMIT_PARAM = "limit"
|
|
|
|
|
|
class CursorDecodeError(ValueError):
|
|
"""Raised when the ``cursor`` query param cannot be decoded."""
|
|
|
|
|
|
def encode_cursor(values: Dict[str, Any]) -> str:
|
|
# ``default=str`` handles datetimes (ISO) so keyset cursors can carry
|
|
# datetime-typed sort fields (e.g. finished_at for match history).
|
|
raw = json.dumps(values, separators=(",", ":"), default=str).encode("utf-8")
|
|
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
|
|
|
|
|
|
def decode_cursor(s: Optional[str]) -> Optional[Dict[str, Any]]:
|
|
if s is None or s == "":
|
|
return None
|
|
try:
|
|
# Tolerate missing padding.
|
|
padded = s + "=" * (-len(s) % 4)
|
|
raw = base64.urlsafe_b64decode(padded.encode("ascii"))
|
|
obj = json.loads(raw.decode("utf-8"))
|
|
except (ValueError, UnicodeDecodeError) as exc:
|
|
raise CursorDecodeError("invalid cursor") from exc
|
|
if not isinstance(obj, dict):
|
|
raise CursorDecodeError("invalid cursor")
|
|
return obj
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Cursor:
|
|
limit: int
|
|
after: Optional[Dict[str, Any]]
|
|
|
|
|
|
def parse_cursor_params(query_string: str) -> Cursor:
|
|
params = extract_query_params(query_string)
|
|
limit_raw = params.get(LIMIT_PARAM)
|
|
if limit_raw:
|
|
try:
|
|
limit = int(limit_raw[0])
|
|
except ValueError as exc:
|
|
raise CursorDecodeError("invalid limit") from exc
|
|
else:
|
|
limit = DEFAULT_LIMIT
|
|
limit = max(MIN_LIMIT, min(MAX_LIMIT, limit))
|
|
after = decode_cursor(params.get(CURSOR_PARAM, [None])[0])
|
|
return Cursor(limit=limit, after=after)
|
|
|
|
|
|
# A sort field: (model field name, "ASC" or "DESC"). The tuple is the full
|
|
# keyset; the cursor encodes exactly these fields.
|
|
Sort = List[Tuple[str, str]]
|
|
|
|
|
|
def _keyset_where(sort: Sort, after: Dict[str, Any]):
|
|
"""Build a Tortoise ``Q`` filter from a cursor."""
|
|
from tortoise.queryset import Q # local to keep import edge narrow
|
|
|
|
clauses = []
|
|
for i, (field, direction) in enumerate(sort):
|
|
key = f"{field}__{'gt' if direction == 'ASC' else 'lt'}"
|
|
value = after.get(field)
|
|
if value is None:
|
|
return Q()
|
|
clause = Q(**{key: value})
|
|
for j in range(i):
|
|
prev_field, _ = sort[j]
|
|
prev_value = after.get(prev_field)
|
|
if prev_value is None:
|
|
return Q()
|
|
clause = clause & Q(**{prev_field: prev_value})
|
|
clauses.append(clause)
|
|
result = clauses[0]
|
|
for clause in clauses[1:]:
|
|
result = result | clause
|
|
return result
|
|
|
|
|
|
async def paginate(
|
|
queryset: QuerySet,
|
|
sort: Sort,
|
|
cursor: Cursor,
|
|
) -> Tuple[List[Any], Optional[str]]:
|
|
"""Return one page of ``queryset`` plus the opaque cursor to continue."""
|
|
order_by: List[str] = []
|
|
for field, direction in sort:
|
|
order_by.append(field if direction == "ASC" else f"-{field}")
|
|
qs = queryset.order_by(*order_by)
|
|
if cursor.after:
|
|
qs = qs.filter(_keyset_where(sort, cursor.after))
|
|
rows = await qs.limit(cursor.limit + 1)
|
|
if len(rows) <= cursor.limit:
|
|
return rows, None
|
|
page = rows[: cursor.limit]
|
|
last = page[-1]
|
|
key: Dict[str, Any] = {}
|
|
for field, _ in sort:
|
|
key[field] = getattr(last, field)
|
|
return page, encode_cursor(key)
|