Add extra info to exceptions raised in pipelines. Fixes #407

ResponseErrors generated by commands executed in a pipeline now includes the
command position in the pipeline and the actual command sent to the Redis
server. For example:

Command # 3 (LPUSH c 3) of pipeline caused error: <actual error message from
    Redis server>

5 files changed, 78 insertions, 26 deletions

diff --git a/CHANGES b/CHANGES
index 89f2d7b..3100531 100644
--- a/CHANGES
+++ b/CHANGES
@@ -11,6 +11,9 @@
     * Error messages that the server sends to the client are now included
       in the client error message. Thanks Sangjin Lim.
     * Added the SCAN, SSCAN, HSCAN, and ZSCAN commands. Thanks Jingchao Hu.
+    * ResponseErrors generated by pipeline execution provide addition context
+      including the position of the command in the pipeline and the actual
+      command text generated the error.
 * 2.8.0
     * redis-py should play better with gevent when a gevent Timeout is raised.
       Thanks leifkb.
diff --git a/redis/client.py b/redis/client.py
index 2820886..4071d32 100644
--- a/redis/client.py
+++ b/redis/client.py
@@ -4,8 +4,9 @@ import datetime
 import sys
 import warnings
 import time as mod_time
-from redis._compat import (b, izip, imap, iteritems, iterkeys, itervalues,
-                           basestring, long, nativestr, urlparse, bytes)
+from redis._compat import (b, basestring, bytes, imap, iteritems, iterkeys,
+                           itervalues, izip, long, nativestr, urlparse,
+                           unicode)
 from redis.connection import ConnectionPool, UnixDomainSocketConnection
 from redis.exceptions import (
     ConnectionError,
@@ -2034,11 +2035,13 @@ class BasePipeline(object):
             errors.append((0, sys.exc_info()[1]))
 
         # and all the other commands
-        for i, _ in enumerate(commands):
+        for i, command in enumerate(commands):
             try:
                 self.parse_response(connection, '_')
             except ResponseError:
-                errors.append((i, sys.exc_info()[1]))
+                ex = sys.exc_info()[1]
+                self.annotate_exception(ex, i + 1, command[0])
+                errors.append((i, ex))
 
         # parse the EXEC.
         try:
@@ -2063,7 +2066,7 @@ class BasePipeline(object):
 
         # find any errors in the response and raise if necessary
         if raise_on_error:
-            self.raise_first_error(response)
+            self.raise_first_error(commands, response)
 
         # We have to run response callbacks manually
         data = []
@@ -2092,14 +2095,21 @@ class BasePipeline(object):
                 response.append(sys.exc_info()[1])
 
         if raise_on_error:
-            self.raise_first_error(response)
+            self.raise_first_error(commands, response)
         return response
 
-    def raise_first_error(self, response):
-        for r in response:
+    def raise_first_error(self, commands, response):
+        for i, r in enumerate(response):
             if isinstance(r, ResponseError):
+                self.annotate_exception(r, i + 1, commands[i][0])
                 raise r
 
+    def annotate_exception(self, exception, number, command):
+        cmd = unicode(' ').join(imap(unicode, command))
+        msg = unicode('Command # %d (%s) of pipeline caused error: %s') % (
+            number, cmd, unicode(exception.args[0]))
+        exception.args = (msg,) + exception.args[1:]
+
     def parse_response(self, connection, command_name, **options):
         result = StrictRedis.parse_response(
             self, connection, command_name, **options)
diff --git a/redis/connection.py b/redis/connection.py
index 9127b13..62682d0 100644
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -29,11 +29,7 @@ SYM_LF = b('\n')
 SYM_EMPTY = b('')
 
 
-class PythonParser(object):
-    "Plain Python parsing class"
-    MAX_READ_LENGTH = 1000000
-    encoding = None
-
+class BaseParser(object):
     EXCEPTION_CLASSES = {
         'ERR': ResponseError,
         'EXECABORT': ExecAbortError,
@@ -41,6 +37,20 @@ class PythonParser(object):
         'NOSCRIPT': NoScriptError,
     }
 
+    def parse_error(self, response):
+        "Parse an error response"
+        error_code = response.split(' ')[0]
+        if error_code in self.EXCEPTION_CLASSES:
+            response = response[len(error_code) + 1:]
+            return self.EXCEPTION_CLASSES[error_code](response)
+        return ResponseError(response)
+
+
+class PythonParser(BaseParser):
+    "Plain Python parsing class"
+    MAX_READ_LENGTH = 1000000
+    encoding = None
+
     def __init__(self):
         self._fp = None
 
@@ -94,14 +104,6 @@ class PythonParser(object):
             raise ConnectionError("Error while reading from socket: %s" %
                                   (e.args,))
 
-    def parse_error(self, response):
-        "Parse an error response"
-        error_code = response.split(' ')[0]
-        if error_code in self.EXCEPTION_CLASSES:
-            response = response[len(error_code) + 1:]
-            return self.EXCEPTION_CLASSES[error_code](response)
-        return ResponseError(response)
-
     def read_response(self):
         response = self.read()
         if not response:
@@ -149,7 +151,7 @@ class PythonParser(object):
         return response
 
 
-class HiredisParser(object):
+class HiredisParser(BaseParser):
     "Parser class for connections using Hiredis"
     def __init__(self):
         if not HIREDIS_AVAILABLE:
@@ -194,6 +196,8 @@ class HiredisParser(object):
             if not buffer.endswith(SYM_LF):
                 continue
             response = self._reader.gets()
+        if isinstance(response, ResponseError):
+            response = self.parse_error(response.args[0])
         return response
 
 if HIREDIS_AVAILABLE:
diff --git a/redis/exceptions.py b/redis/exceptions.py
index d67afa7..0b64f91 100644
--- a/redis/exceptions.py
+++ b/redis/exceptions.py
@@ -1,10 +1,21 @@
 "Core exceptions raised by the Redis client"
+from redis._compat import unicode
 
 
 class RedisError(Exception):
     pass
 
 
+# python 2.5 doesn't implement Exception.__unicode__. Add it here to all
+# our exception types
+if not hasattr(RedisError, '__unicode__'):
+    def __unicode__(self):
+        if isinstance(self.args[0], unicode):
+            return self.args[0]
+        return unicode(self.args[0])
+    RedisError.__unicode__ = __unicode__
+
+
 class AuthenticationError(RedisError):
     pass
 
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index ee860e7..46fc994 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -2,7 +2,7 @@ from __future__ import with_statement
 import pytest
 
 import redis
-from redis._compat import b
+from redis._compat import b, u, unichr, unicode
 
 
 class TestPipeline(object):
@@ -105,8 +105,10 @@ class TestPipeline(object):
         r['c'] = 'a'
         with r.pipeline() as pipe:
             pipe.set('a', 1).set('b', 2).lpush('c', 3).set('d', 4)
-            with pytest.raises(redis.ResponseError):
+            with pytest.raises(redis.ResponseError) as ex:
                 pipe.execute()
+            assert unicode(ex.value).startswith('Command # 3 (LPUSH c 3) of '
+                                                'pipeline caused error: ')
 
             # make sure the pipe was restored to a working state
             assert pipe.set('z', 'zzz').execute() == [True]
@@ -116,9 +118,12 @@ class TestPipeline(object):
         with r.pipeline() as pipe:
             # the zrem is invalid because we don't pass any keys to it
             pipe.set('a', 1).zrem('b').set('b', 2)
-            with pytest.raises(redis.ResponseError):
+            with pytest.raises(redis.ResponseError) as ex:
                 pipe.execute()
 
+            assert unicode(ex.value).startswith('Command # 2 (ZREM b) of '
+                                                'pipeline caused error: ')
+
             # make sure the pipe was restored to a working state
             assert pipe.set('z', 'zzz').execute() == [True]
             assert r['z'] == b('zzz')
@@ -196,7 +201,26 @@ class TestPipeline(object):
             pipe.llen('a')
             pipe.expire('a', 100)
 
-            with pytest.raises(redis.ResponseError):
+            with pytest.raises(redis.ResponseError) as ex:
                 pipe.execute()
 
+            assert unicode(ex.value).startswith('Command # 1 (LLEN a) of '
+                                                'pipeline caused error: ')
+
         assert r['a'] == b('1')
+
+    def test_exec_error_in_no_transaction_pipeline_unicode_command(self, r):
+        key = unichr(3456) + u('abcd') + unichr(3421)
+        r[key] = 1
+        with r.pipeline(transaction=False) as pipe:
+            pipe.llen(key)
+            pipe.expire(key, 100)
+
+            with pytest.raises(redis.ResponseError) as ex:
+                pipe.execute()
+
+            expected = unicode('Command # 1 (LLEN %s) of pipeline caused '
+                               'error: ') % key
+            assert unicode(ex.value).startswith(expected)
+
+        assert r[key] == b('1')