HEX
Server: Apache/2.4.52 (Ubuntu)
System: Linux spn-python 5.15.0-89-generic #99-Ubuntu SMP Mon Oct 30 20:42:41 UTC 2023 x86_64
User: arjun (1000)
PHP: 8.1.2-1ubuntu2.20
Disabled: NONE
Upload Files
File: //home/arjun/projects/aigenerator/venv/lib64/python3.12/site-packages/django_redis/hash_ring.py
import bisect
import hashlib
from typing import Dict, Iterable, Iterator, List, Optional, Tuple


class HashRing:
    nodes: List[str] = []

    def __init__(self, nodes: Iterable[str] = (), replicas: int = 128) -> None:
        self.replicas: int = replicas
        self.ring: Dict[str, str] = {}
        self.sorted_keys: List[str] = []

        for node in nodes:
            self.add_node(node)

    def add_node(self, node: str) -> None:
        self.nodes.append(node)

        for x in range(self.replicas):
            _key = f"{node}:{x}"
            _hash = hashlib.sha256(_key.encode()).hexdigest()

            self.ring[_hash] = node
            self.sorted_keys.append(_hash)

        self.sorted_keys.sort()

    def remove_node(self, node: str) -> None:
        self.nodes.remove(node)
        for x in range(self.replicas):
            _hash = hashlib.sha256(f"{node}:{x}".encode()).hexdigest()
            del self.ring[_hash]
            self.sorted_keys.remove(_hash)

    def get_node(self, key: str) -> Optional[str]:
        n, i = self.get_node_pos(key)
        return n

    def get_node_pos(self, key: str) -> Tuple[Optional[str], Optional[int]]:
        if len(self.ring) == 0:
            return None, None

        _hash = hashlib.sha256(key.encode()).hexdigest()
        idx = bisect.bisect(self.sorted_keys, _hash)
        idx = min(idx - 1, (self.replicas * len(self.nodes)) - 1)
        return self.ring[self.sorted_keys[idx]], idx

    def iter_nodes(self, key: str) -> Iterator[Tuple[Optional[str], Optional[str]]]:
        if len(self.ring) == 0:
            yield None, None

        node, pos = self.get_node_pos(key)
        for k in self.sorted_keys[pos:]:
            yield k, self.ring[k]

    def __call__(self, key: str) -> Optional[str]:
        return self.get_node(key)