Repository: redis/redis-py
File: redis/cluster.py (Python)

Code before:
        return self

    def raise_first_error(self, stack):
        """
        Raise the first exception on the stack
        """
        for c in stack:
            r = c.result
            if isinstance(r, Exception):
                self.annotate_exception(r, c.position + 1, c.args)
                raise r

    def annotate_exception(self, exception, number, command):
        """
        Provides extra context to the exception prior to it being handled
        """
        cmd = " ".join(map(safe_str, command))
        msg = (
            f"Command # {number} ({self._truncate_command(cmd)}) of pipeline "
            f"caused error: {exception.args[0]}"
        )
        exception.args = (msg,) + exception.args[1:]

    def _truncate_command(self, command, max_length=100):
        if len(command) > max_length:
            return command[: max_length - 3] + "x.."
        return command

    def execute(self, raise_on_error: bool = True) -> List[Any]:
        """
        Execute all the commands in the current pipeline
        """
        stack = self.command_stack
        try:
            return self.send_cluster_commands(stack, raise_on_error)
        finally:
            self.reset()

    def reset(self):
        """
        Reset back to empty pipeline.
        """
        self.command_stack = []

        self.scripts = set()

        # TODO: Implement
        # make sure to reset the connection state in the event that we were
        # watching something
        # if self.watching and self.connection:
        #     try:

Review comment (suggestion):
@rohansingh  Here is the problem with tests - you are not adding '...', instead 'x..' gets in the truncated string.

Code after:

    def __init__(
        self,
        nodes_manager: "NodesManager",
        commands_parser: "CommandsParser",
        result_callbacks: Optional[Dict[str, Callable]] = None,
        cluster_response_callbacks: Optional[Dict[str, Callable]] = None,
        startup_nodes: Optional[List["ClusterNode"]] = None,
        read_from_replicas: bool = False,
        load_balancing_strategy: Optional[LoadBalancingStrategy] = None,
        cluster_error_retry_attempts: int = 3,
        reinitialize_steps: int = 5,
        lock=None,
        **kwargs,
    ):
        """ """
        self.command_stack = []
        self.nodes_manager = nodes_manager
        self.commands_parser = commands_parser
        self.refresh_table_asap = False
        self.result_callbacks = (
            result_callbacks or self.__class__.RESULT_CALLBACKS.copy()
        )
        self.startup_nodes = startup_nodes if startup_nodes else []
        self.read_from_replicas = read_from_replicas
        self.load_balancing_strategy = load_balancing_strategy
        self.command_flags = self.__class__.COMMAND_FLAGS.copy()
        self.cluster_response_callbacks = cluster_response_callbacks
        self.cluster_error_retry_attempts = cluster_error_retry_attempts
        self.reinitialize_counter = 0
        self.reinitialize_steps = reinitialize_steps
        self.encoder = Encoder(
            kwargs.get("encoding", "utf-8"),
            kwargs.get("encoding_errors", "strict"),
            kwargs.get("decode_responses", False),
        )
        if lock is None:
            lock = threading.Lock()
        self._lock = lock

    def __repr__(self):
        """ """
        return f"{type(self).__name__}"

    def __enter__(self):
        """ """
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        """ """
        self.reset()