Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Common (?) pattern for consuming paged objects as a stream in Python, using Stripe Subscriptions as an example
import functools
import stripe
from typing import Any, Optional
def stripe_subscriptions(limit: Optional[int] = 100, **params: Any):
"""
Returns an iterator of all matching subscriptions
This abstracts away the need to deal with pages of results
or items, and simply get a stream (which can be turned into
a list, for example) of subscriptions.
Args:
limit: the number of subscriptions per page; default is
set high because we assume
**params: any valid Stripe Subscription list params
Returns:
a generator of individual subscriptions
"""
# By using partial we ensure that when this function call is repeated we're
# reusing all of the same base arguments. It's a minor benefit, to be sure,
# but eliminates a small class of fat finger errors writing or updating
# the function.
query = functools.partial(stripe.Subscription.list, limit=limit, **params)
results = query()
while results:
# This the only thing we want to consume, a stream of stripe.Subscription objects
for sub in results:
yield sub
if results.has_more:
# Pagination is accomplished by asking for items after a
# given Stripe ID, which would be the last item. The
# results are iterable but not indexable, which is why
# the data attribute (the underlying list of subscriptions)
# is used here.
results = query(starting_after=results.data[-1].id)
else:
return
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment