Skip to content

Instantly share code, notes, and snippets.

@epoz
Last active August 29, 2015 14:03
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save epoz/f4a1024d89616df6fac5 to your computer and use it in GitHub Desktop.
Save epoz/f4a1024d89616df6fac5 to your computer and use it in GitHub Desktop.
Elasticsearch Python client adaptor to be used with Django Paginator
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment