omnigent-ai--omnigent
wehub-resource-sync
106 行
3.3 KiB
Python
106 行
3.3 KiB
Python
"""Cursor-based pagination container."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
from typing import Generic, TypeVar
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
@dataclass
|
|
class PagedList(Generic[T]):
|
|
"""
|
|
A page of results matching the OpenAI list pagination shape.
|
|
|
|
:param data: The items in this page.
|
|
:param first_id: ID of the first item in the page, or ``None``
|
|
if empty.
|
|
:param last_id: ID of the last item in the page, or ``None``
|
|
if empty.
|
|
:param has_more: ``True`` if more pages exist after this one.
|
|
"""
|
|
|
|
data: list[T] = field(default_factory=list)
|
|
first_id: str | None = None
|
|
last_id: str | None = None
|
|
has_more: bool = False
|
|
|
|
|
|
def paginate_in_memory(
|
|
items: list[T],
|
|
id_fn: Callable[[T], str],
|
|
*,
|
|
limit: int = 20,
|
|
after: str | None = None,
|
|
before: str | None = None,
|
|
order: str = "desc",
|
|
) -> PagedList[T]:
|
|
"""Apply cursor-based pagination to an in-memory list.
|
|
|
|
Items should already be in a stable order. The function slices
|
|
using ``after``/``before`` cursors keyed by item id and returns
|
|
at most ``limit`` items.
|
|
|
|
Forward pagination (``after``) returns up to ``limit`` items
|
|
immediately after the cursor. Backward pagination (``before``)
|
|
returns up to ``limit`` items immediately *before* the cursor,
|
|
anchored to the end of the range — not the first page. ``has_more``
|
|
reports whether more items remain in the pagination direction:
|
|
after the page for ``after``, before the page for ``before``. An
|
|
unknown cursor id is ignored.
|
|
|
|
:param items: Pre-sorted items to paginate.
|
|
:param id_fn: Callable that extracts the id from an item.
|
|
:param limit: Maximum items to return, default 20.
|
|
:param after: Cursor id — return items after this one.
|
|
:param before: Cursor id — return items before this one.
|
|
:param order: ``"desc"`` reverses the list, ``"asc"`` keeps it.
|
|
:returns: A paginated :class:`PagedList`.
|
|
"""
|
|
working = list(items)
|
|
if order == "desc":
|
|
working = list(reversed(working))
|
|
|
|
# Resolve the cursors to a ``[start, end)`` window over ``working``.
|
|
# An unknown cursor leaves its bound untouched (it is ignored).
|
|
start = 0
|
|
end = len(working)
|
|
|
|
if after is not None:
|
|
idx = next(
|
|
(i for i, item in enumerate(working) if id_fn(item) == after),
|
|
None,
|
|
)
|
|
if idx is not None:
|
|
start = idx + 1
|
|
|
|
before_found = False
|
|
if before is not None:
|
|
idx = next(
|
|
(i for i, item in enumerate(working) if id_fn(item) == before),
|
|
None,
|
|
)
|
|
if idx is not None:
|
|
end = idx
|
|
before_found = True
|
|
|
|
if before_found:
|
|
# Backward: the ``limit`` items immediately preceding the cursor,
|
|
# anchored to the end of the window (not the first page).
|
|
page_start = max(start, end - limit)
|
|
page = working[page_start:end]
|
|
has_more = page_start > start
|
|
else:
|
|
# Forward, or no/unknown cursor: the first ``limit`` items.
|
|
page = working[start:end][:limit]
|
|
has_more = end - start > limit
|
|
|
|
return PagedList(
|
|
data=page,
|
|
first_id=id_fn(page[0]) if page else None,
|
|
last_id=id_fn(page[-1]) if page else None,
|
|
has_more=has_more,
|
|
)
|