mirror of
https://github.com/stack-auth/stack.git
synced 2026-07-20 21:29:36 +08:00
feat: add pagination support with cursor-based PaginatedResult
- PaginatedResult[T] generic wrapping nested {items, pagination: {next_cursor}} shape
- Convenience properties .next_cursor and .has_next_page
- Package __init__.py exports all models, errors, and PaginatedResult
- SyncAPIClient/AsyncAPIClient intentionally excluded from public API
- Tests verify pagination with ServerUser and Team types
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
"""Stack Auth Python SDK."""
|
||||
|
||||
from stack_auth._pagination import PaginatedResult
|
||||
from stack_auth._version import __version__
|
||||
from stack_auth.errors import (
|
||||
AnalyticsError,
|
||||
@@ -17,9 +18,40 @@ from stack_auth.errors import (
|
||||
StackAuthError,
|
||||
ValidationError,
|
||||
)
|
||||
from stack_auth.models import (
|
||||
ActiveSession,
|
||||
ApiKey,
|
||||
BaseUser,
|
||||
ContactChannel,
|
||||
GeoInfo,
|
||||
Item,
|
||||
NotificationCategory,
|
||||
OAuthConnection,
|
||||
OAuthProvider,
|
||||
Product,
|
||||
Project,
|
||||
ProjectConfig,
|
||||
ProjectPermission,
|
||||
ServerTeam,
|
||||
ServerUser,
|
||||
Team,
|
||||
TeamApiKey,
|
||||
TeamApiKeyFirstView,
|
||||
TeamInvitation,
|
||||
TeamMemberProfile,
|
||||
TeamPermission,
|
||||
UserApiKey,
|
||||
UserApiKeyFirstView,
|
||||
)
|
||||
|
||||
# NOTE: SyncAPIClient and AsyncAPIClient are intentionally NOT exported here.
|
||||
# They are internal transport classes in _client.py.
|
||||
# Phase 3 will add the public StackServerApp and AsyncStackServerApp facades.
|
||||
|
||||
__all__ = [
|
||||
"__version__",
|
||||
"PaginatedResult",
|
||||
# Errors
|
||||
"StackAuthError",
|
||||
"AuthenticationError",
|
||||
"NotFoundError",
|
||||
@@ -34,4 +66,28 @@ __all__ = [
|
||||
"RateLimitError",
|
||||
"CliError",
|
||||
"AnalyticsError",
|
||||
# Models
|
||||
"BaseUser",
|
||||
"ServerUser",
|
||||
"Team",
|
||||
"ServerTeam",
|
||||
"TeamMemberProfile",
|
||||
"TeamInvitation",
|
||||
"ActiveSession",
|
||||
"GeoInfo",
|
||||
"ContactChannel",
|
||||
"TeamPermission",
|
||||
"ProjectPermission",
|
||||
"Project",
|
||||
"ProjectConfig",
|
||||
"ApiKey",
|
||||
"UserApiKey",
|
||||
"UserApiKeyFirstView",
|
||||
"TeamApiKey",
|
||||
"TeamApiKeyFirstView",
|
||||
"OAuthConnection",
|
||||
"OAuthProvider",
|
||||
"Product",
|
||||
"Item",
|
||||
"NotificationCategory",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
"""Cursor-based pagination wrapper for Stack Auth list API responses."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Generic, TypeVar
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
class _PaginationMeta(BaseModel):
|
||||
"""Nested pagination metadata from API response."""
|
||||
|
||||
next_cursor: str | None = None
|
||||
|
||||
|
||||
class PaginatedResult(BaseModel, Generic[T]):
|
||||
"""Cursor-based pagination wrapper for list API responses.
|
||||
|
||||
Matches the Stack Auth API response shape:
|
||||
{ "items": [...], "pagination": { "next_cursor": "..." } }
|
||||
|
||||
Exposes .next_cursor and .has_next_page as top-level convenience properties.
|
||||
"""
|
||||
|
||||
items: list[T]
|
||||
pagination: _PaginationMeta = Field(default_factory=_PaginationMeta)
|
||||
|
||||
@property
|
||||
def next_cursor(self) -> str | None:
|
||||
"""The cursor for fetching the next page, or None if no more pages."""
|
||||
return self.pagination.next_cursor
|
||||
|
||||
@property
|
||||
def has_next_page(self) -> bool:
|
||||
"""Whether there are more pages available."""
|
||||
return self.pagination.next_cursor is not None
|
||||
@@ -0,0 +1,112 @@
|
||||
"""Tests for PaginatedResult[T] generic pagination wrapper."""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class TestPaginatedResult:
|
||||
"""Tests for PaginatedResult with nested pagination shape."""
|
||||
|
||||
def test_parse_with_server_user(self) -> None:
|
||||
from stack_auth._pagination import PaginatedResult
|
||||
from stack_auth.models.users import ServerUser
|
||||
|
||||
data = {
|
||||
"items": [
|
||||
{
|
||||
"id": "u1",
|
||||
"signedUpAtMillis": 1711296000000,
|
||||
"hasPassword": True,
|
||||
"otpAuthEnabled": False,
|
||||
"passkeyAuthEnabled": False,
|
||||
"isMultiFactorRequired": False,
|
||||
"isAnonymous": False,
|
||||
"isRestricted": False,
|
||||
}
|
||||
],
|
||||
"pagination": {"next_cursor": "abc123"},
|
||||
}
|
||||
result = PaginatedResult[ServerUser].model_validate(data)
|
||||
assert len(result.items) == 1
|
||||
assert isinstance(result.items[0], ServerUser)
|
||||
assert result.items[0].id == "u1"
|
||||
assert result.next_cursor == "abc123"
|
||||
|
||||
def test_no_next_cursor_means_no_next_page(self) -> None:
|
||||
from stack_auth._pagination import PaginatedResult
|
||||
from stack_auth.models.teams import Team
|
||||
|
||||
data = {
|
||||
"items": [{"id": "t1", "displayName": "Team A"}],
|
||||
"pagination": {},
|
||||
}
|
||||
result = PaginatedResult[Team].model_validate(data)
|
||||
assert result.next_cursor is None
|
||||
assert result.has_next_page is False
|
||||
|
||||
def test_next_cursor_present_means_has_next_page(self) -> None:
|
||||
from stack_auth._pagination import PaginatedResult
|
||||
from stack_auth.models.teams import Team
|
||||
|
||||
data = {
|
||||
"items": [{"id": "t1", "displayName": "Team A"}],
|
||||
"pagination": {"next_cursor": "cursor123"},
|
||||
}
|
||||
result = PaginatedResult[Team].model_validate(data)
|
||||
assert result.has_next_page is True
|
||||
|
||||
def test_empty_items_list(self) -> None:
|
||||
from stack_auth._pagination import PaginatedResult
|
||||
from stack_auth.models.users import BaseUser
|
||||
|
||||
data = {"items": [], "pagination": {}}
|
||||
result = PaginatedResult[BaseUser].model_validate(data)
|
||||
assert len(result.items) == 0
|
||||
assert result.has_next_page is False
|
||||
|
||||
def test_items_count(self) -> None:
|
||||
from stack_auth._pagination import PaginatedResult
|
||||
from stack_auth.models.teams import Team
|
||||
|
||||
data = {
|
||||
"items": [
|
||||
{"id": "t1", "displayName": "Team A"},
|
||||
{"id": "t2", "displayName": "Team B"},
|
||||
{"id": "t3", "displayName": "Team C"},
|
||||
],
|
||||
"pagination": {"next_cursor": "next"},
|
||||
}
|
||||
result = PaginatedResult[Team].model_validate(data)
|
||||
assert len(result.items) == 3
|
||||
|
||||
def test_item_is_correct_type(self) -> None:
|
||||
from stack_auth._pagination import PaginatedResult
|
||||
from stack_auth.models.users import ServerUser
|
||||
|
||||
data = {
|
||||
"items": [
|
||||
{
|
||||
"id": "u1",
|
||||
"signedUpAtMillis": 1711296000000,
|
||||
"hasPassword": False,
|
||||
"otpAuthEnabled": False,
|
||||
"passkeyAuthEnabled": False,
|
||||
"isMultiFactorRequired": False,
|
||||
"isAnonymous": False,
|
||||
"isRestricted": False,
|
||||
"serverMetadata": {},
|
||||
}
|
||||
],
|
||||
"pagination": {},
|
||||
}
|
||||
result = PaginatedResult[ServerUser].model_validate(data)
|
||||
assert isinstance(result.items[0], ServerUser)
|
||||
|
||||
def test_pagination_absent_uses_default(self) -> None:
|
||||
"""If pagination key is missing entirely, default to no next page."""
|
||||
from stack_auth._pagination import PaginatedResult
|
||||
from stack_auth.models.teams import Team
|
||||
|
||||
data = {"items": [{"id": "t1", "displayName": "Team A"}]}
|
||||
result = PaginatedResult[Team].model_validate(data)
|
||||
assert result.next_cursor is None
|
||||
assert result.has_next_page is False
|
||||
Reference in New Issue
Block a user