Method get_node_from_slot

redis/cluster.py:2260–2299

Gets a node that servers this hash slot

(
        self,
        slot: int,
        read_from_replicas: bool = False,
        load_balancing_strategy: Optional[LoadBalancingStrategy] = None,
        server_type: Optional[Literal["primary", "replica"]] = None,
    )

        version="5.3.0",
    )
    def get_node_from_slot(
        self,
        slot: int,
        read_from_replicas: bool = False,
        load_balancing_strategy: Optional[LoadBalancingStrategy] = None,
        server_type: Optional[Literal["primary", "replica"]] = None,
    ) -> ClusterNode:
        """
        Gets a node that servers this hash slot
        """

        if read_from_replicas is True and load_balancing_strategy is None:
            load_balancing_strategy = LoadBalancingStrategy.ROUND_ROBIN

        with self._lock:
            if self.slots_cache.get(slot) is None or len(self.slots_cache[slot]) == 0:
                raise SlotNotCoveredError(
                    f'Slot "{slot}" not covered by the cluster. '
                    + f'"require_full_coverage={self._require_full_coverage}"'
                )

            if len(self.slots_cache[slot]) > 1 and load_balancing_strategy:
                # get the server index using the strategy defined in load_balancing_strategy
                primary_name = self.slots_cache[slot][0].name
                node_idx = self.read_load_balancer.get_server_index(
                    primary_name, len(self.slots_cache[slot]), load_balancing_strategy
                )
            elif (
                server_type is None
                or server_type == PRIMARY
                or len(self.slots_cache[slot]) == 1
            ):
                # return a primary
                node_idx = 0
            else:
                # return a replica
                # randomly choose one of the replicas
                node_idx = random.randint(1, len(self.slots_cache[slot]) - 1)

            return self.slots_cache[slot][node_idx]

    def get_nodes_by_server_type(self, server_type: Literal["primary", "replica"]):
        """