A generator is a Python iterable that produces values incrementally instead of materializing them all at once. In an API data pipeline, it can hide pagination and expose individual rating records through one interface. section 9.1.2
A paginated generator separates fetching mechanics from the code consuming records:
- Request a page with offset and limit parameters. section 9.1.2
- Raise an error for unsuccessful responses and decode the JSON payload. section 9.1.2
- Use yield from to expose each record from the current page. section 9.1.2
- Fetch the next page only after the current page’s records have been exhausted. section 9.1.2
Listing 9.3 Helper function for handling pagination
def _get_with_pagination(session, url, params, batch_size=100):
"""
Fetches records using a GET request with given URL/params,
taking pagination into account.
"""
offset = 0 #1
total = None #1
while total is None or offset < total: #2
response = session.get(
url,
params={
**params,
**{"offset": offset, "limit": batch_size} #3
}
)
response.raise_for_status()
response_json = response.json() #4
yield from response_json["result"] #5
offset += batch_size #6
total = response_json["total"] #6
A pagination helper yields records while tracking offsets, totals, response validation, and page size. section 9.1.2
Checking whether any ratings exist
A custom Airflow sensor can call next on the generator with batch_size=1. If a record is produced, the sensor returns True; if the generator is empty and raises StopIteration, it returns False and waits for another check. section 9.4
Listing 9.29 Implementing the poke method
class MovielensRatingsSensor(BaseSensorOperator):
def poke(self, context:Context):
hook = MovielensHook(self._conn_id)
try:
next( #1
hook.get_ratings(
start_date=self._start_date,
end_date=self._end_date,
batch_size=1
)
)
self.log.info(
f"Found ratings for {self._start_date} to {self._end_date}"
)
return True #2
except StopIteration: #3
self.log.info(
f"Didn't find any ratings for {self._start_date} "
f"to {self._end_date}, waiting..."
)
return False #3
finally:
hook.close() #4
The sensor implementation demonstrates generator consumption, StopIteration handling, logging, and hook cleanup. section 9.4
