1# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
  2
  3from typing import Any, List, Generic, TypeVar, Optional, cast
  4from typing_extensions import Protocol, override, runtime_checkable
  5
  6from ._base_client import BasePage, PageInfo, BaseSyncPage, BaseAsyncPage
  7
  8__all__ = [
  9    "SyncPage",
 10    "AsyncPage",
 11    "SyncCursorPage",
 12    "AsyncCursorPage",
 13    "SyncConversationCursorPage",
 14    "AsyncConversationCursorPage",
 15]
 16
 17_T = TypeVar("_T")
 18
 19
 20@runtime_checkable
 21class CursorPageItem(Protocol):
 22    id: Optional[str]
 23
 24
 25class SyncPage(BaseSyncPage[_T], BasePage[_T], Generic[_T]):
 26    """Note: no pagination actually occurs yet, this is for forwards-compatibility."""
 27
 28    data: List[_T]
 29    object: str
 30
 31    @override
 32    def _get_page_items(self) -> List[_T]:
 33        data = self.data
 34        if not data:
 35            return []
 36        return data
 37
 38    @override
 39    def next_page_info(self) -> None:
 40        """
 41        This page represents a response that isn't actually paginated at the API level
 42        so there will never be a next page.
 43        """
 44        return None
 45
 46
 47class AsyncPage(BaseAsyncPage[_T], BasePage[_T], Generic[_T]):
 48    """Note: no pagination actually occurs yet, this is for forwards-compatibility."""
 49
 50    data: List[_T]
 51    object: str
 52
 53    @override
 54    def _get_page_items(self) -> List[_T]:
 55        data = self.data
 56        if not data:
 57            return []
 58        return data
 59
 60    @override
 61    def next_page_info(self) -> None:
 62        """
 63        This page represents a response that isn't actually paginated at the API level
 64        so there will never be a next page.
 65        """
 66        return None
 67
 68
 69class SyncCursorPage(BaseSyncPage[_T], BasePage[_T], Generic[_T]):
 70    data: List[_T]
 71    has_more: Optional[bool] = None
 72
 73    @override
 74    def _get_page_items(self) -> List[_T]:
 75        data = self.data
 76        if not data:
 77            return []
 78        return data
 79
 80    @override
 81    def has_next_page(self) -> bool:
 82        has_more = self.has_more
 83        if has_more is not None and has_more is False:
 84            return False
 85
 86        return super().has_next_page()
 87
 88    @override
 89    def next_page_info(self) -> Optional[PageInfo]:
 90        data = self.data
 91        if not data:
 92            return None
 93
 94        item = cast(Any, data[-1])
 95        if not isinstance(item, CursorPageItem) or item.id is None:
 96            # TODO emit warning log
 97            return None
 98
 99        return PageInfo(params={"after": item.id})
100
101
102class AsyncCursorPage(BaseAsyncPage[_T], BasePage[_T], Generic[_T]):
103    data: List[_T]
104    has_more: Optional[bool] = None
105
106    @override
107    def _get_page_items(self) -> List[_T]:
108        data = self.data
109        if not data:
110            return []
111        return data
112
113    @override
114    def has_next_page(self) -> bool:
115        has_more = self.has_more
116        if has_more is not None and has_more is False:
117            return False
118
119        return super().has_next_page()
120
121    @override
122    def next_page_info(self) -> Optional[PageInfo]:
123        data = self.data
124        if not data:
125            return None
126
127        item = cast(Any, data[-1])
128        if not isinstance(item, CursorPageItem) or item.id is None:
129            # TODO emit warning log
130            return None
131
132        return PageInfo(params={"after": item.id})
133
134
135class SyncConversationCursorPage(BaseSyncPage[_T], BasePage[_T], Generic[_T]):
136    data: List[_T]
137    has_more: Optional[bool] = None
138    last_id: Optional[str] = None
139
140    @override
141    def _get_page_items(self) -> List[_T]:
142        data = self.data
143        if not data:
144            return []
145        return data
146
147    @override
148    def has_next_page(self) -> bool:
149        has_more = self.has_more
150        if has_more is not None and has_more is False:
151            return False
152
153        return super().has_next_page()
154
155    @override
156    def next_page_info(self) -> Optional[PageInfo]:
157        last_id = self.last_id
158        if not last_id:
159            return None
160
161        return PageInfo(params={"after": last_id})
162
163
164class AsyncConversationCursorPage(BaseAsyncPage[_T], BasePage[_T], Generic[_T]):
165    data: List[_T]
166    has_more: Optional[bool] = None
167    last_id: Optional[str] = None
168
169    @override
170    def _get_page_items(self) -> List[_T]:
171        data = self.data
172        if not data:
173            return []
174        return data
175
176    @override
177    def has_next_page(self) -> bool:
178        has_more = self.has_more
179        if has_more is not None and has_more is False:
180            return False
181
182        return super().has_next_page()
183
184    @override
185    def next_page_info(self) -> Optional[PageInfo]:
186        last_id = self.last_id
187        if not last_id:
188            return None
189
190        return PageInfo(params={"after": last_id})