-
Notifications
You must be signed in to change notification settings - Fork 5
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Allow pagination to be auto or manual #87
Comments
Hi @jibranmazhar. Unfortunately airlines can be quite slow! We are looking at ways to improve the experience though. Behind-the-scenes, When you're calling |
@jesse-c no i did not set |
I see, I understand now! That's the approach we take in our Ruby client. I'm discussing this with my colleagues about for having an option for auto-pagination or not. We'll be triaging the prioritisation of this improvement. |
I implemented something like this so I can use the after/before parameters: from typing import Any, Generic, List, Optional, TypeVar
from pydantic import BaseModel
ClientT = TypeVar("ClientT", bound="HttpClient")
CallerT = TypeVar("CallerT", bound=BaseModel)
class PaginationMetadata(BaseModel):
limit: int
after: Optional[str]
class PaginatedResult(BaseModel, Generic[CallerT]):
data: List[CallerT]
meta: Optional[PaginationMetadata]
class Pagination(Generic[ClientT, CallerT]):
"""A way to do pagination on list() calls"""
def __init__(self, client: ClientT, caller: CallerT, params: dict[str, Any]):
self._client = client
self._caller = caller
if params["limit"] > 200:
# We're vaguely faking the structure of the error structure returned
# from the API.
raise ApiError([], {"errors": [{"message": "limit exceeds 200"}]})
self._params = params
async def __aiter__(self):
"""Iterate over the response items and yield one by one"""
response = await self.get()
while "meta" in response:
after = response["meta"]["after"]
for entry in response["data"]:
yield self._caller.from_json(entry)
if after is None:
break
response = await self.get(after)
async def execute(self, *, after=None, before=None) -> PaginatedResult[CallerT]:
response = await self.get(after=after, before=before)
if response is None:
return PaginatedResult(data=[], meta=None)
return PaginatedResult(
data=[self._caller.model_validate(offer) for offer in response["data"]],
meta=PaginationMetadata(**response["meta"]),
)
async def get(self, *, after=None, before=None):
try:
if after:
self._params["after"] = after
self._params.pop("before")
elif before:
self._params["before"] = before
self._params.pop("after")
except KeyError:
pass
return await self._client.do_get(
self._client._url,
query_params=self._params,
) Which can then be used like so: async def list_offers(list_offers_input):
return await duffel_client.offers.list(
list_offers_input.offer_request_id,
sort=list_offers_input.sort.name,
limit=list_offers_input.limit,
max_connections=list_offers_input.max_connections,
).execute(after=list_offers_input.after, before=list_offers_input.before) |
Hello
i got a problem in duffel.offers.list() call because it take almost 11 sec to give data. i explain the scenario first user create offer request with all required data with this api ( duffel.offer_request.execute() without return_offer() ) so this call take 3 to 7 sec and after that i go to the next page with offer ID and make another request to get offers with ( duffel.offers.list() ) and this call take 9 to 12 sec because they got all offers at once with pagination which i cannot control so please give control in pagination in api call because pagination model call api recursively and its stop when all offer they get thank you i hope you understand what i mean.
The text was updated successfully, but these errors were encountered: