Source code for linode_api4.paginated_list

import math


[docs]class PaginatedList(object): """ The PaginatedList encapsulates the API V4's pagination in an easily consumable way. A PaginatedList may be treated like a normal `list` in all ways, and can be iterated over, indexed, and sliced. PaginatedLists should never be constructed manually, and instead should be created by requesting a collection of resources from the :any:`LinodeClient`. For example:: linodes = client.linode.instances() # returns a PaginatedList of Linodes Once you have a PaginatedList of resources, it doesn't matter how many resources the API will return - you can iterate over all of them without having to worry about pagination.:: # iterate over all linodes. If there are two or more pages, # they will be loaded as required. for linode in linodes: print(linode.label) You may access the number of items in a collection by calling `len` on the PaginatedList:: num_linodes = len(linodes) This will _not_ emit another API request. """ def __init__(self, client, page_endpoint, page=[], max_pages=1, total_items=None, parent_id=None, filters=None): self.client = client self.page_endpoint = page_endpoint self.query_filters = filters self.page_size = len(page) self.max_pages = max_pages self.lists = [ None for _ in range(0, self.max_pages) ] if self.lists: self.lists[0] = page self.list_cls = type(page[0]) if page else None # TODO if this is None that's bad self.objects_parent_id = parent_id self.cur = 0 # for being a generator self.total_items = total_items if not total_items: self.total_items = len(page)
[docs] def first(self): """ A convenience method for getting only the first item in this list. Exactly equivalent to getting index 0. :returns: The first item in this list. """ return self[0]
[docs] def last(self): """ A convenience method for getting only the last item in this list. Exactly equivalent to getting index -1. :returns: The first item in this list. """ return self[-1]
[docs] def only(self): """ Returns the first item in this list, and asserts that it is the only item. This is useful when querying a collection for a resource and expecting to get only one back. For instance:: # raises if it finds more than one Linode production_box = client.linode.instances(Linode.group == "prod").only() :returns: The first and only item in this list. :raises ValueError: If more than one item is in this list. """ if len(self) == 1: return self[0] raise ValueError("List {} has more than one element!".format(self))
def __repr__(self): return "PaginatedList ({} items)".format(self.total_items) def _load_page(self, page_number): j = self.client.get("/{}?page={}&page_size={}".format(self.page_endpoint, page_number+1, self.page_size), filters=self.query_filters) if j['pages'] != self.max_pages or j['results'] != len(self): raise RuntimeError('List {} has changed since creation!'.format(self)) l = PaginatedList.make_list(j["data"], self.client, self.list_cls, parent_id=self.objects_parent_id) self.lists[page_number] = l def __getitem__(self, index): # this comes in here now, but we're hadling it elsewhere if isinstance(index, slice): return self._get_slice(index) # handle negative indexing if index < 0: index = len(self) + index if index < 0: raise IndexError('list index out of range') if index >= self.page_size * self.max_pages: raise IndexError('list index out of range') normalized_index = index % self.page_size target_page = math.ceil((index+1.0)/self.page_size)-1 target_page = int(target_page) if not self.lists[target_page]: self._load_page(target_page) return self.lists[target_page][normalized_index] def __len__(self): return self.total_items def _get_slice(self, s): # get range i = s.start if s.start is not None else 0 j = s.stop if s.stop is not None else self.total_items # we do not support steps outside of 1 yet if s.step is not None and s.step != 1: raise NotImplementedError('Only step sizes of 1 are currently supported.') # if i or j are negative, normalize them if i < 0: i = self.total_items + i if j < 0: j = self.total_items + j # if i or j are still negative, that's an IndexError if i < 0 or j < 0: raise IndexError('list index out of range') # if we're going nowhere or backward, return nothing if j <= i: return [] result = [] for c in range(i, j): result.append(self[c]) return result def __setitem__(self, index, value): raise AttributeError('Assigning to indicies in paginated lists is not supported') def __delitem__(self, index): raise AttributeError('Deleting from paginated lists is not supported') def __next__(self): if self.cur < len(self): self.cur += 1 return self[self.cur-1] else: raise StopIteration() @staticmethod def make_list(json_arr, client, cls, parent_id=None): """ Returns a list of Populated objects of the given class type. This should not be called outside of the :any:`LinodeClient` class. :param json_arr: The array of JSON data to make into a list :param client: The LinodeClient to pass to new objects :param parent_id: The parent id for derived objects :returns: A list of models from the JSON """ result = [] for obj in json_arr: id_val = None if 'id' in obj: id_val = obj['id'] elif hasattr(cls, 'id_attribute') and getattr(cls, 'id_attribute') in obj: id_val = obj[getattr(cls, 'id_attribute')] else: continue o = cls.make_instance(id_val, client, parent_id=parent_id, json=obj) result.append(o) return result @staticmethod def make_paginated_list(json, client, cls, parent_id=None, page_url=None, filters=None): """ Returns a PaginatedList populated with the first page of data provided, and the ability to load additional pages. This should not be called outside of the :any:`LinodeClient` class. :param json: The JSON list to use as the first page :param client: A LinodeClient to use to load additional pages :param parent_id: The parent ID for derived objects :param page_url: The URL to use when loading more pages :param cls: The class to instantiate for objects :param filters: The filters used when making the call that generated this list. If not provided, this will fail when loading additional pages. :returns: An instance of PaginatedList that will represent the entire collection whose first page is json """ l = PaginatedList.make_list(json["data"], client, cls, parent_id=parent_id) p = PaginatedList(client, page_url, page=l, max_pages=json['pages'], total_items=json['results'], parent_id=parent_id, filters=filters) return p