The story of real-time ranking using Redis Sorted Set has already been messed up, but I felt like I searched quickly, and I could not find a concrete implementation example in Python, so I usually simplify the library that I use in-house. I will introduce it.
Source Code
ranking.py
from datetime import timedelta
from .rankinglist import RankingList
_default_expire = int(timedelta(weeks=2).total_seconds())
class Ranking(object):
def __init__(self, client, key, expire=_default_expire):
"""
Initialize the class.
:param object client:Redis client.
:param string key:Ranking identifier.
:param int expire:Ranking expiration date.Two weeks if omitted.
"""
self._r = client
self._key = key
self._expire = expire
def push(self, unique_id, value):
"""
Update the ranking.
:param string unique_id:ID to rank.
:param string value:Ranking source value(Points etc.)
"""
self._r.zadd(self._key, long(value), unique_id)
self.touch()
def get_rank(self, unique_id):
"""
Get the ranking.
:param string unique_id:ID to rank.
:return:Ranking
"""
value = self._r.zscore(self._key, unique_id)
if value is None:
return None
return self._r.zcount(self._key, '({}'.format(int(value)), 'inf') + 1
def get_range(self, start, end):
"""
Get the ranking range.
:param int start:Start position of subscript.
:param int end:End position of subscript. start=0 and end=At 0,Get the first one.
:return: ['push()Unique specified in_id', ...]
"""
result = self._r.zrevrange(self._key, start, end)
self.touch()
return result
def get_count(self):
"""
Get the number of cases.
:return:number
"""
return self._r.zcard(self._key)
def touch(self):
"""
Extend the expiration date of the ranking.
push()And get_rank()But it will be executed.
"""
self._r.expire(self._key, self._expire)
def clean(self):
"""
Delete ranking.
"""
self._r.delete(self._key)
def gen_list(self, wrapper=None):
"""
Get a ranking list.
:param function wrapper:Function that wraps the element
:return:RankingList object
RankingList,Act as a List to some extent.
Used when passing to Django's Paginator, etc..
"""
return RankingList(self, wrapper)
A little confusing are get_rank () and gen_list (). gen_rank () counts the number of people higher than the current score in order to obtain a ranking that takes into account the ties. The RankingList object returned by gen_list () will be described later.
rankinglist.py
class RankingList(object):
def __init__(self, rank, wrapper=None):
self._rank = rank
self._wrapper = wrapper
def __getitem__(self, k):
if isinstance(k, slice):
start = k.start if k.start else 0
end = k.stop - 1 if k.stop else self.__len__() - 1
step = k.step
unique_ids = self._rank.get_range(start, end)
if step:
unique_ids = unique_ids[::step]
return [self._wrap(unique_id) for unique_id in unique_ids]
else:
if self.__len__() <= k:
raise IndexError('list index out of range')
unique_ids = self._rank.get_range(k, k)
return self._wrap(unique_ids[0])
def _wrap(self, unique_id):
return self._wrapper(unique_id) if self._wrapper else unique_id
def __len__(self):
return self._rank.get_count()
Delegate Ranking to a RankingList that behaves like a Python List. Also, by passing Wapper, Object is generated and returned based on the unique_id extracted from Redis.
>>> from redis import Redis
>>> from ranking import Ranking
>>>
>>> ranking = Ranking(Redis(), 'event1')
>>>
>>> ranking.push('p1', 200)
>>> ranking.push('p2', 100)
>>> ranking.push('p3', 300)
>>> ranking.push('p1', 1000)
>>> ranking.push('p4', 1000)
>>>
>>> l1 = ranking.gen_list() # ['p4', 'p1', 'p3', 'p2']
>>> l1[2:] # ['p3', 'p2']
>>>
>>> import Player # e.g. Django Model
>>> def wrapper(id):
return Player.objects.get(pk=id)
>>> l2 = ranking.gen_list(wrapper) # [Player('p4'), Player('p1'), Player('p3'), Player('p2')]
>>> l2[2:] # [Player('p3'), Player('p2')]
>>>
>>> [ranking.get_rank(player_id) for player_id in l1] # [1, 1, 2, 3]
Recommended Posts