In converting one of my projects from the old pyes client to the new official Python client for Elasticsearch, I needed to support the Django paginator as it is used in various bits of my existing code.
After some (unsuccessful) searching for existing implementations, this is what I came up with. Simple if you see it written out like this, but it took me the good part of an evening and morning to do. Sigh, what a slowpoke!
class ElasticSearchPaginatorListException(Exception):
pass
class ElasticSearchPaginatorList(object):
def __init__(self, client, *args, **kwargs):
self.client = client
self.args = args
self.kwargs = kwargs
self._count = None
def count(self):
if self._count is None:
result = self.client.count(index=self.kwargs['index'],
body=self.kwargs['body'])
return result['count']
def __len__(self):
return self.count()
def __getitem__(self, key):
if not isinstance(key, slice):
raise ElasticSearchPaginatorListException('key parameter in __getitem__ is not a slice instance')
self.kwargs['from_'] = key.start
self.kwargs['size'] = key.stop - key.start
return self.client.search(*self.args, **self.kwargs)['hits']['hits']
if __name__ == '__main__':
import elasticsearch
from django.core.paginator import Paginator
e = ElasticSearchPaginatorList(elasticsearch.Elasticsearch(),
index='_all',
body={'query': {'match_all': {}}
})
first_list, second_list = [], []
p = Paginator(e, 5)
for x in p.page(1):
first_list.append(x['_id'])
for x in p.page(2):
first_list.append(x['_id'])
p = Paginator(e, 10)
for x in p.page(1):
second_list.append(x['_id'])
assert(first_list == second_list)