fix: Read pagination from the response instead of a client hook

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y1RzepycXEYA3LStfjt8cY

Author: claude

codegen/layouts/partials/route-method.hbs

@@ -35,7 +35,10 @@
         return None
 {{else if (isListType returnType)}}
 
-        return [{{fromDict (listItemType returnType)}}(item) for item in unwrap_list(res, "{{returnPath.[0]}}", "{{path}}")]
+        return {{#if hasPagination}}PaginatedList(
+            [{{fromDict (listItemType returnType)}}(item) for item in unwrap_list(res, "{{returnPath.[0]}}", "{{path}}")],
+            pagination=res.get("pagination"),
+        ){{else}}[{{fromDict (listItemType returnType)}}(item) for item in unwrap_list(res, "{{returnPath.[0]}}", "{{path}}")]{{/if}}
 {{else}}
 
         return {{fromDict returnType}}(unwrap(res, "{{returnPath.[0]}}", "{{path}}"))

codegen/layouts/route.hbs

@@ -20,6 +20,9 @@ from ..response import unwrap
 {{#if importUnwrapList}}
 from ..response import unwrap_list
 {{/if}}
+{{#if importPaginatedList}}
+from ..pagination import PaginatedList
+{{/if}}
 
 
 {{> abstract-route-class abstractClass}}

codegen/lib/layouts/route.ts

@@ -64,6 +64,7 @@ export interface RouteLayoutContext {
   importNull: boolean
   importUnwrap: boolean
   importUnwrapList: boolean
+  importPaginatedList: boolean
   methods: MethodLayoutContext[]
 }
 
@@ -154,6 +155,8 @@ export const setRouteLayoutContext = (cls: ClassModel): RouteLayoutContext => {
       returnPath.length > 0 && returnType.startsWith('List['),
   )
 
+  const importPaginatedList = methods.some(({ hasPagination }) => hasPagination)
+
   const showPass =
     cls.methods.length === 0 && cls.childClassIdentifiers.length === 0
 
@@ -198,6 +201,7 @@ export const setRouteLayoutContext = (cls: ClassModel): RouteLayoutContext => {
     importNull,
     importUnwrap,
     importUnwrapList,
+    importPaginatedList,
     methods,
   }
 }

seam/pagination.py

@@ -1,3 +1,6 @@
+from typing import Any, Dict, List, Optional
+
+
 class Pagination:
     def __init__(
         self,
@@ -8,3 +11,16 @@ def __init__(
         self.has_next_page = has_next_page
         self.next_page_cursor = next_page_cursor
         self.next_page_url = next_page_url
+
+
+class PaginatedList(List[Any]):
+    """A list of results that carries the response's pagination envelope.
+
+    Behaves exactly like the plain list it replaces; the paginator reads
+    the ``pagination`` attribute instead of intercepting the response
+    through a client-wide event hook.
+    """
+
+    def __init__(self, items: List[Any], pagination: Optional[Dict[str, Any]] = None):
+        super().__init__(items)
+        self.pagination = pagination

seam/paginator.py

@@ -8,9 +8,8 @@
     Optional,
     Tuple,
 )
-from json import JSONDecodeError
-from httpx import Response
 from .client import AsyncSeamHttpClient, SeamHttpClient
+from .exceptions import SeamHttpInvalidResponseError
 from .pagination import Pagination
 
 
@@ -22,15 +21,29 @@ def parse_pagination(pagination: Dict[str, Any]) -> Pagination:
     )
 
 
+def read_pagination(data: Any, request: Callable) -> Pagination:
+    """Read the pagination envelope a paginated route attaches to its result."""
+
+    pagination = getattr(data, "pagination", None)
+
+    if not isinstance(pagination, dict):
+        path = getattr(request, "__seam_path__", "this endpoint")
+        raise SeamHttpInvalidResponseError(
+            path,
+            "pagination",
+            f"got {type(pagination).__name__} instead of a pagination object",
+        )
+
+    return parse_pagination(pagination)
+
+
 class SeamPaginator:
     """
     Handles pagination for API list endpoints.
 
     Iterates through pages of results returned by a callable function.
     """
 
-    _FIRST_PAGE = "FIRST_PAGE"
-
     def __init__(
         self,
         client: SeamHttpClient,
@@ -48,19 +61,12 @@ def __init__(
         self._request = request
         self.client = client
         self._params = params or {}
-        self._pagination_cache: Dict[str, Pagination] = {}
 
     def first_page(self) -> Tuple[List[Any], Pagination | None]:
         """Fetches the first page of results."""
-        self.client.event_hooks["response"].append(
-            lambda response: self._cache_pagination(response, self._FIRST_PAGE)
-        )
         data = self._request(**self._params)
-        self.client.event_hooks["response"].pop()
 
-        pagination = self._pagination_cache.get(self._FIRST_PAGE)
-
-        return data, pagination
+        return data, read_pagination(data, self._request)
 
     def next_page(
         self, next_page_cursor: str, /
@@ -74,15 +80,9 @@ def next_page(
             "page_cursor": next_page_cursor,
         }
 
-        self.client.event_hooks["response"].append(
-            lambda response: self._cache_pagination(response, next_page_cursor)
-        )
         data = self._request(**params)
-        self.client.event_hooks["response"].pop()
 
-        pagination = self._pagination_cache.get(next_page_cursor)
-
-        return data, pagination
+        return data, read_pagination(data, self._request)
 
     def flatten_to_list(self) -> List[Any]:
         """Fetches all pages and returns all items as a single list."""
@@ -110,18 +110,6 @@ def flatten(self) -> Generator[Any, None, None]:
             if current_items:
                 yield from current_items
 
-    def _cache_pagination(self, response: Response, page_key: str) -> None:
-        """Extracts pagination dict from response, creates Pagination object, and caches it."""
-        try:
-            # httpx response hooks fire before the response body is read.
-            response.read()
-            pagination = response.json().get("pagination", {})
-        except JSONDecodeError:
-            pagination = {}
-
-        if isinstance(pagination, dict):
-            self._pagination_cache[page_key] = parse_pagination(pagination)
-
 
 class AsyncSeamPaginator:
     """
@@ -130,8 +118,6 @@ class AsyncSeamPaginator:
     Iterates through pages of results returned by an awaitable function.
     """
 
-    _FIRST_PAGE = "FIRST_PAGE"
-
     def __init__(
         self,
         client: AsyncSeamHttpClient,
@@ -149,21 +135,12 @@ def __init__(
         self._request = request
         self.client = client
         self._params = params or {}
-        self._pagination_cache: Dict[str, Pagination] = {}
 
     async def first_page(self) -> Tuple[List[Any], Pagination | None]:
         """Fetches the first page of results."""
-
-        async def cache_pagination(response: Response) -> None:
-            await self._cache_pagination(response, self._FIRST_PAGE)
-
-        self.client.event_hooks["response"].append(cache_pagination)
         data = await self._request(**self._params)
-        self.client.event_hooks["response"].pop()
-
-        pagination = self._pagination_cache.get(self._FIRST_PAGE)
 
-        return data, pagination
+        return data, read_pagination(data, self._request)
 
     async def next_page(
         self, next_page_cursor: str, /
@@ -177,16 +154,9 @@ async def next_page(
             "page_cursor": next_page_cursor,
         }
 
-        async def cache_pagination(response: Response) -> None:
-            await self._cache_pagination(response, next_page_cursor)
-
-        self.client.event_hooks["response"].append(cache_pagination)
         data = await self._request(**params)
-        self.client.event_hooks["response"].pop()
 
-        pagination = self._pagination_cache.get(next_page_cursor)
-
-        return data, pagination
+        return data, read_pagination(data, self._request)
 
     async def flatten_to_list(self) -> List[Any]:
         """Fetches all pages and returns all items as a single list."""
@@ -217,15 +187,3 @@ async def flatten(self) -> AsyncGenerator[Any, None]:
             )
             for item in current_items or []:
                 yield item
-
-    async def _cache_pagination(self, response: Response, page_key: str) -> None:
-        """Extracts pagination dict from response, creates Pagination object, and caches it."""
-        try:
-            # httpx response hooks fire before the response body is read.
-            await response.aread()
-            pagination = response.json().get("pagination", {})
-        except JSONDecodeError:
-            pagination = {}
-
-        if isinstance(pagination, dict):
-            self._pagination_cache[page_key] = parse_pagination(pagination)