Implement ranking processing with ties in Python using Redis Sorted Set

Origin

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 processing

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.

Treat the ranking as a list

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.

How to use

>>> 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

Implement ranking processing with ties in Python using Redis Sorted Set
Implement Redis Mutex in Python
Using Python mode in Processing
Easy image processing in Python with Pillow
Sorted list in Python
Implement Enigma in python
File processing in Python
Multithreaded processing in python
Implement recommendations in Python
Implement XENO in python
Queue processing in Python
Image processing with Python
Process csv data with python (count processing using pandas)
Implement sum in Python
Implement Traceroute in Python 3
Things to keep in mind when using Python with AtCoder
Things to keep in mind when using cgi with python.
Morphological analysis using Igo + mecab-ipadic-neologd in Python (with Ruby bonus)
Image processing with Python (Part 2)
100 Language Processing with Python Knock 2015
UTF8 text processing in python
Scraping with selenium in Python
"Apple processing" with OpenCV3 + Python3
Working with LibreOffice in Python
Scraping with chromedriver in python
Using Quaternion with Python ~ numpy-quaternion ~
Debugging with pdb in Python
Acoustic signal processing with Python (2)
[Python] Using OpenCV with Python (Basic)
Acoustic signal processing with Python
Working with sounds in Python
Scraping with Selenium in Python
Asynchronous processing (threading) in python
Implement naive bayes in Python 3.3
Implement ancient ciphers in python
Image processing with Python (Part 1)
Scraping with Tor in Python
Tweet with image in Python
Combined with permutations in Python
Image processing with Python (Part 3)
Redis pipe lining in Python
Implement fast RPC in Python
Implement method chain in Python
Image Processing Collection in Python
Implement Dijkstra's Algorithm in python
Implement Slack chatbot in Python
Implement reversi reversi processing using BitBoard
Translate using googletrans in Python
Using OpenCV with Python @Mac
Send using Python with Gmail
Set python test in jenkins
[Python] Image processing with scikit-image
Median filter using xarray (median filter)
Image processing with Python 100 knock # 10 median filter
I get a can't set attribute when using @property in python
Object extraction in images by pattern matching using OpenCV with Python
Complement python with emacs using company-jedi
Number recognition in images with Python
Handle Base91 keys with python + redis.
Harmonic mean with Python Harmonic mean (using SciPy)
GUI programming in Python using Appjar
Implement R's power.prop.test function in python