-
Notifications
You must be signed in to change notification settings - Fork 143
Expand file tree
/
Copy pathcustom_pagination.py
More file actions
152 lines (121 loc) · 5.11 KB
/
custom_pagination.py
File metadata and controls
152 lines (121 loc) · 5.11 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
# This file was auto-generated by Fern from our API Definition.
"""
Custom Pagination Support
This file is designed to be modified by SDK users to implement their own
pagination logic. The generator will import SyncCustomPager and AsyncCustomPager
from this module when custom pagination is used.
Users should:
1. Implement their custom pager (e.g., PayrocPager, MyCustomPager, etc.)
2. Create adapter classes (SyncCustomPager/AsyncCustomPager) that bridge
between the generated SDK code and their custom pager implementation
"""
from __future__ import annotations
from typing import Any, AsyncIterator, Generic, Iterator, TypeVar
# Import the base utilities you'll need
# Adjust these imports based on your actual structure
try:
from .client_wrapper import AsyncClientWrapper, SyncClientWrapper
except ImportError:
# Fallback for type hints
AsyncClientWrapper = Any # type: ignore
SyncClientWrapper = Any # type: ignore
TItem = TypeVar("TItem")
TResponse = TypeVar("TResponse")
class SyncCustomPager(Generic[TItem, TResponse]):
"""
Adapter for custom synchronous pagination.
The generator will call this with:
SyncCustomPager(initial_response=response, client_wrapper=client_wrapper)
Implement this class to extract pagination metadata from your response
and delegate to your custom pager implementation.
Example implementation:
class SyncCustomPager(Generic[TItem, TResponse]):
def __init__(
self,
*,
initial_response: TResponse,
client_wrapper: SyncClientWrapper,
):
# Extract data and pagination metadata from response
data = initial_response.data # Adjust based on your response structure
links = initial_response.links
# Initialize your custom pager
self._pager = MyCustomPager(
current_page=Page(data),
httpx_client=client_wrapper.httpx_client,
get_headers=client_wrapper.get_headers,
# ... other parameters
)
def __iter__(self):
return iter(self._pager)
# Delegate other methods to your pager...
"""
def __init__(
self,
*,
initial_response: TResponse,
client_wrapper: SyncClientWrapper,
):
"""
Initialize the custom pager.
Args:
initial_response: The parsed API response from the first request
client_wrapper: The client wrapper providing HTTP client and utilities
"""
raise NotImplementedError(
"SyncCustomPager must be implemented. "
"Please implement this class in core/custom_pagination.py to define your pagination logic. "
"See the class docstring for examples."
)
def __iter__(self) -> Iterator[TItem]:
"""Iterate through all items across all pages."""
raise NotImplementedError("Must implement __iter__ method")
class AsyncCustomPager(Generic[TItem, TResponse]):
"""
Adapter for custom asynchronous pagination.
The generator will call this with:
AsyncCustomPager(initial_response=response, client_wrapper=client_wrapper)
Implement this class to extract pagination metadata from your response
and delegate to your custom async pager implementation.
Example implementation:
class AsyncCustomPager(Generic[TItem, TResponse]):
def __init__(
self,
*,
initial_response: TResponse,
client_wrapper: AsyncClientWrapper,
):
# Extract data and pagination metadata from response
data = initial_response.data # Adjust based on your response structure
links = initial_response.links
# Initialize your custom async pager
self._pager = MyAsyncCustomPager(
current_page=Page(data),
httpx_client=client_wrapper.httpx_client,
get_headers=client_wrapper.get_headers,
# ... other parameters
)
async def __aiter__(self):
return self._pager.__aiter__()
# Delegate other methods to your pager...
"""
def __init__(
self,
*,
initial_response: TResponse,
client_wrapper: AsyncClientWrapper,
):
"""
Initialize the custom async pager.
Args:
initial_response: The parsed API response from the first request
client_wrapper: The client wrapper providing HTTP client and utilities
"""
raise NotImplementedError(
"AsyncCustomPager must be implemented. "
"Please implement this class in core/custom_pagination.py to define your pagination logic. "
"See the class docstring for examples."
)
async def __aiter__(self) -> AsyncIterator[TItem]:
"""Asynchronously iterate through all items across all pages."""
raise NotImplementedError("Must implement __aiter__ method")