bvandeusenandClaude Opus 5.5
7275fad767
#5211 "Lists load as you scroll: audit, alert history, Ansible runs" (milestone 477), per preference 172. - core/paging.py: keyset paging over (time, id), newest first, with an opaque "<iso>|<id>" cursor; a page reads one row past itself to know whether another follows. No OFFSET, so rows added while someone scrolls never shift the pages after them. - _macros.html scroll_more / scroll_end and static/scroll-more.js, written once: an IntersectionObserver with a 600px margin fetches the next page before the reader reaches the end; "Loading more…" while in flight, "End of the list." at the end, and a failed page says so with a Retry button instead of re-firing on every scroll. - Audit log: no "Load more" link and no 500-row cap. - Alert history: no 200-row stop with "pick a shorter range"; alert_svc.history takes the cursor, so the MCP tool and the page share one query. - Ansible runs: no silent cap of 50. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
"""Keyset paging for newest-first lists that load as you scroll (#5211, pref 172).
|
|
|
|
A page is the rows strictly older than a cursor, ordered (time, id) newest
|
|
first, so a row added while someone scrolls never shifts the pages after it,
|
|
and no page needs an OFFSET. The cursor is "<iso time>|<id>": opaque to the
|
|
browser, which only hands back the one the last page gave it.
|
|
|
|
stmt = keyset(select(AuditEvent), AuditEvent.timestamp, AuditEvent.id, request.args.get("before"))
|
|
rows = (await db.execute(stmt)).scalars().all()
|
|
rows, after = next_page(rows, lambda e: (e.timestamp, e.id))
|
|
|
|
Templates end each page with scroll_more(<url with ?before=after>) while
|
|
`after` is set, and scroll_end() once it is not (_macros.html).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Callable
|
|
|
|
from sqlalchemy import tuple_
|
|
|
|
# Rows per page: enough to fill a tall screen past the scroll margin, so the
|
|
# next page starts loading before the reader reaches the end of this one.
|
|
PAGE_SIZE = 100
|
|
|
|
|
|
def cursor(at: datetime, row_id) -> str:
|
|
return f"{at.isoformat()}|{row_id}"
|
|
|
|
|
|
def parse_cursor(raw: str | None) -> tuple[datetime, str] | None:
|
|
"""(time, id) from a cursor, or None for no cursor or one that does not parse
|
|
(a mangled cursor starts the list over rather than failing the page)."""
|
|
if not raw or "|" not in raw:
|
|
return None
|
|
at, row_id = raw.split("|", 1)
|
|
try:
|
|
return datetime.fromisoformat(at), row_id
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def keyset(stmt, at_col, id_col, before: str | None, size: int = PAGE_SIZE):
|
|
"""`stmt` limited to one page older than `before`, newest first. Fetches
|
|
one row more than a page so next_page can tell whether another follows."""
|
|
parsed = parse_cursor(before)
|
|
if parsed is not None:
|
|
stmt = stmt.where(tuple_(at_col, id_col) < tuple_(*parsed))
|
|
return stmt.order_by(at_col.desc(), id_col.desc()).limit(size + 1)
|
|
|
|
|
|
def next_page(rows, key: Callable, size: int = PAGE_SIZE) -> tuple[list, str | None]:
|
|
"""(this page's rows, the cursor for the next page or None at the end).
|
|
`key(row)` is the row's (time, id)."""
|
|
rows = list(rows)
|
|
if len(rows) <= size:
|
|
return rows, None
|
|
rows = rows[:size]
|
|
return rows, cursor(*key(rows[-1]))
|