Implémenter le traitement du classement avec des liens en Python à l'aide de Redis Sorted Set

Origine

L'histoire du classement en temps réel à l'aide de l'ensemble trié Redis a déjà été effacée, mais cela ressemble à une recherche rapide, et je n'ai pas pu trouver d'exemple d'implémentation concret en Python, donc je simplifie généralement la bibliothèque que j'utilise en interne Je vais le présenter.

Source Code

Traitement du classement

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):
        """
Initialiser la classe.

        :param object client:Client Redis.
        :param string key:Identifiant de classement.
        :param int expire:Date d'expiration du classement.Deux semaines si omis.
        """
        self._r = client
        self._key = key
        self._expire = expire

    def push(self, unique_id, value):
        """
Mettre à jour le classement.

        :param string unique_id:ID à classer.
        :param string value:Valeur de la source de classement(Points etc.)
        """
        self._r.zadd(self._key, long(value), unique_id)
        self.touch()

    def get_rank(self, unique_id):
        """
Obtenez le classement.

        :param string unique_id:ID à classer.
        :return:Classement
        """
        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):
        """
Obtenez la plage de classement.

        :param int start:Position de départ de l'indice.
        :param int end:Position finale de l'indice. start=0 and end=À 0,Obtenez le premier.
        :return: ['push()Unique spécifié dans_id', ...]
        """
        result = self._r.zrevrange(self._key, start, end)
        self.touch()
        return result

    def get_count(self):
        """
Obtenez le nombre de cas.

        :return:nombre
        """
        return self._r.zcard(self._key)

    def touch(self):
        """
Prolonger la date d'expiration du classement.
        push()Et obtenir_rank()Mais il sera exécuté.
        """
        self._r.expire(self._key, self._expire)

    def clean(self):
        """
Supprimer le classement.
        """
        self._r.delete(self._key)

    def gen_list(self, wrapper=None):
        """
Obtenez la liste de classement.

        :param function wrapper:Fonction qui enveloppe l'élément
        :return:Objet RankingList

ClassementListe,Agir comme une liste dans une certaine mesure.
Utilisé lors du passage à Paginator de Django etc..
        """
        return RankingList(self, wrapper)

Un peu déroutant sont get_rank () et gen_list (). gen_rank () compte le nombre de personnes supérieur au score actuel afin d'obtenir un classement prenant en compte les égalités. L'objet RankingList retourné par gen_list () sera décrit plus tard.

Traitez le classement comme une liste

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

Déléguez le classement à une RankingList qui se comporte comme une liste Python. De plus, en passant Wapper, un objet est généré et renvoyé en fonction de l'identifiant unique extrait de Redis.

Comment utiliser

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